S3FC project page

S3FC home page

---

               

---

s3_message.cc

Go to the documentation of this file.

/*
 * Stone Three Foundation Class (s3fc) provides a number of utility classes.
 * Copyright (C) 2001 by Stone Three Signal Processing (Pty) Ltd.
 *
 * Authored by Stone Three Signal Processing (Pty) Ltd.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 * 
 * Please see the file 'COPYING' in the source root directory.
 */

#include <s3fc/s3_message.h>
#include <s3fc/s3_exception.h>
#include <s3fc/s3_config_node.h>
#include <sstream>
#include <algorithm>

S3_DEFINE_MAGIC_STRING(s3_message);

// dftl ctor - empty message
s3_message::s3_message() :
   empty(true),
   from(""),
   to(""),
   body(""),
   mc(NORMAL),
   id(""),
   reply_id("")
{
}

// init ctor - never empty
s3_message::s3_message(const std::string& n_from,
             const std::string& n_to,
             const std::string& n_body,
             const msg_class_t& n_mc,
             const msg_id_t& n_id) :
   empty(false),
   from(n_from),
   to(n_to),
   body(n_body),
   mc(n_mc),
   id(n_id),
   reply_id("")
{
}

//
s3_message::s3_message(const std::string& n_from,
             const s3_message& msg,
             const std::string& n_body,
             const msg_id_t& n_id) :
   empty(false),
   from(n_from),
   to(msg.from),
   body(n_body),
   mc(msg.mc),
   id(n_id),
   reply_id(msg.id)
{
}

// virtual dtor required by compiler
s3_message::~s3_message()
{
}

// pack all fields into buffer 
void s3_message::pack(s3_pack_buffer &b) const
{
   s3_streamable::pack(empty, b);
   s3_streamable::pack(from, b);
   s3_streamable::pack(to, b);
   s3_streamable::pack(body, b);
   s3_streamable::pack(mc, b);
   s3_streamable::pack(id, b);
   s3_streamable::pack(reply_id, b);
}

// unpack all fields from buffer
void s3_message::unpack(const s3_pack_buffer &b) 
{
   s3_streamable::unpack(empty, b);
   s3_streamable::unpack(from, b);
   s3_streamable::unpack(to, b);
   s3_streamable::unpack(body, b);
   s3_streamable::unpack(mc, b);
   s3_streamable::unpack(id, b);
   s3_streamable::unpack(reply_id, b);
}

//
bool s3_message::is_empty() const
{
   return empty;
}

//
bool s3_message::is_reply(const msg_id_t& n_reply_id) const
{
   return (reply_id.size() > 0) && (n_reply_id == reply_id);
}

//
std::string s3_message::get_from() const
{
   return from;
}

//
std::string s3_message::get_to() const
{
   return to;
}

//
std::string s3_message::get_body() const
{
   return body;
}

//
s3_message::msg_class_t s3_message::get_class() const
{
   return mc;
}

//
std::string s3_message::get_debug_str() const
{
   std::stringstream str;
   str << "[ " << from << " -> " << to << " ]";
   switch(mc)
   {
   case NORMAL:
      str << ".";
      break;
   case PRIORITY:
      str << "P";
      break;
   case CONTROL:
      str << "C";
      break;
   default:
      str << " ";
      break;
   }
   if ( reply_id.size() )
   {
      str << "R";
   }
   else if ( is_empty() )
   {
      str << "E";
   }
   else
   {
      str << " ";
   }
   str << " (id: " << id << ") ";
   if ( reply_id.size() )
   {
      str << "(reply_id: " << reply_id << ")";
   }
   if ( ! is_empty() )
   {
      str << std::endl;
      str << "     '" << body << "'";
   }
   else 
   {
      str << " <empty>";
   }
   return str.str();
}

// init the static
s3_post_office* s3_message_box::poffice = 0;
unsigned int s3_message_box::dropped_msg_count = 0;

//
s3_message_box::s3_message_box(const std::string& n_name,
                unsigned int n_rx_queue_len) :
   connected(false),
   name(n_name),
   rx_queue_len(n_rx_queue_len),
   rx_queue(0),
   subscription_list(0),
   seq_nr(0)
{
}

//
s3_message_box::~s3_message_box()
{
   // disconnect if still connected
   if (connected)
   {
      disconnect();
   }
}

//
void s3_message_box::init(const std::string& ip_addr,
           int port)
{
   // test pre-condition
   if ( initialised() )
   {
      throw s3_generic_exception("s3_message_box::init()",
             "Precondition initialised() == false violated");
   }
   poffice = new s3_post_office();
   try
   {
#if S3_HAS_LOCAL_PO_SWITCH
      poffice->connect();
#else
      poffice->connect(ip_addr, port);
#endif
   }
   catch( const s3_exception& e)
   {
      // cleanup and rethrow
      delete poffice;
      poffice = 0;
      throw e;
   }
   
}

//
void s3_message_box::de_init()
{
   if ( ! initialised() )
   {
      throw s3_generic_exception("s3_message_box::de_init()",
             "Precondition initialised() == "
             "true violated");
   }
   // disconnect and stop - this throws if we still have clients
   poffice->disconnect();
   poffice->request_terminate(); // Use soft terminate <jp>
   poffice->wait_on_exit();
   delete poffice;
   poffice = 0;
}

//
bool s3_message_box::initialised()
{
   S3FC_DBG( std::cerr << "s3_message_box::initialised() poffice = "
        << poffice << std::endl << std::flush );

   return (poffice != 0 &&
      poffice->is_connected());
}

//
void s3_message_box::set_name(const std::string& n_name)
{
   if ( connected ) 
   {
      throw s3_generic_exception("s3_message_box::set_name()",
             err_str("Connection exists."));
   }
   name = n_name;
}

//
void s3_message_box::connect()
{
   if ( connected ) 
   {
      throw s3_generic_exception("s3_message_box::connect()",
             err_str("Connection exists."));
   }
   if ( name.empty() )
   {
      std::stringstream what;
      what << "Invalid name '" << name << "'";
      throw s3_generic_exception("s3_message_box::connect()", 
             what.str());
   }
   // register with the post office
   poffice->add_dst(name, *this, false);
   connected = true;
}

//
void s3_message_box::disconnect()
{
   if ( ! connected )
   {
      throw s3_generic_exception("s3_message_box::disconnect()",
             err_str("No connection."));
   }
   // unsubscribe from all groups
   while ( ! subscription_list.empty() )
   {
      unsubscribe_group(subscription_list.front());
      //subscription_list.pop_front();
   }
   // deregister
   poffice->remove_dst(name, *this, false);
   connected = false;
}

//
void s3_message_box::subscribe_group(const std::string& group)
{
   if ( ! connected )
   {
      throw s3_generic_exception("s3_message_box::subscribe_group()",
             err_str("No connection."));
   }
   // see if we're already subscribed
   for ( std::list<std::string>::const_iterator ptr = subscription_list.begin();
    ptr != subscription_list.end();
    ptr++)
   {
      if (*ptr == group)
      {
    std::stringstream str;
    str << "Already subscribed to group: '" << group << "'";
    throw s3_generic_exception("s3_message_box::subscribe_group()",
                err_str(str.str()));
      }
   }
   poffice->add_dst(group, *this, true);
   subscription_list.push_back(group);
}

//
void s3_message_box::unsubscribe_group(const std::string& group)
{
   if ( ! connected )
   {
      throw s3_generic_exception("s3_message_box::unsubscribe_group()",
             err_str("No connection."));
   }
   // check subscription
   std::list<std::string>::iterator ptr;
   ptr = std::find(subscription_list.begin(), subscription_list.end(), group);
   // not found ?
   if ( ptr == subscription_list.end() )
   {
      std::stringstream str;
      str << "Not subscribed to group: '" << group << "'";
      throw s3_generic_exception("s3_message_box::unsubscribe_group()",
             err_str(str.str()));
   }
   poffice->remove_dst(group, *this, true);
   subscription_list.erase(ptr);
}

//
void s3_message_box::subscribe_arrival(const s3_semaphore& sem)
{
   std::list<s3_semaphore* >::const_iterator ptr = 
      std::find(arrival_notification_list.begin(),
      arrival_notification_list.end(),
      &sem);
   if ( ptr != arrival_notification_list.end()) 
   {
      std::stringstream str;
      str << "Already subscribed for arrival: '" << &sem << "'";
      throw s3_generic_exception("s3_message_box::subscribe_arrival()",
             err_str(str.str()));      
   }
   arrival_notification_list.push_back(const_cast<s3_semaphore*>(&sem));
}

//
s3_message::msg_id_t s3_message_box::send_msg(const std::string& dst,
                     const std::string& body,
                     bool priority)
{
   if ( ! connected )
   {
      throw s3_generic_exception("s3_message_box::send_msg()",
             err_str("No connection."));
   }
   s3_message::msg_id_t id = get_next_message_id();
   s3_message msg(name, dst, body, (priority ? 
                s3_message::PRIORITY : 
                s3_message::NORMAL),
        id);
   poffice->dispatch_msg(msg, *this);
   return id;
}

//
s3_message::msg_id_t s3_message_box::reply_msg(const s3_message& msg,
                      const std::string& body)
{
   s3_message::msg_id_t id = get_next_message_id();
   s3_message reply_msg(name, msg, body, id);
   poffice->dispatch_msg(reply_msg, *this);
   return id;
}


// the rx_queue is always sorted candidate-first by the scheme
// implemented in deliver_message
s3_message s3_message_box::get_msg()
{
   // need locks here as we might run concurrently with a post office
   // doing a delivery
   state_lock.lock();

   s3_message msg;
   // candidate is always the first in the queue
   if (! rx_queue.empty() )
   {
      msg = rx_queue.front();
      rx_queue.pop_front();
   }
   else
   {
      // empty message
      msg = s3_message();
   }
   state_lock.unlock();
   return msg;
}

// insert the new message into the queue in such a way that
// the first is always the candidate
void s3_message_box::deliver_msg(const s3_message& msg)
{
   // need locks here as we might run concurrently with a user app
   // doing a get_msg
   state_lock.lock();
   bool rx_changed = false;
   // full queue: remove the oldest non-priority message unless
   //  all queued messages are priority and the new messages is also
   //  priority, in which case the oldest is removed
   if ( rx_queue.size() == rx_queue_len )
   {
      T_RX_Queue::iterator ptr;
      // find oldest (first) non-prio (oldest at beginning)
      for (ptr = rx_queue.begin(); 
      ptr != rx_queue.end();
      ptr++)
      {
    if ((*ptr).get_class() == s3_message::NORMAL) break;
      }
      // no non-prio in queue: remove the oldest prio
      if ( ptr == rx_queue.end() )
      {
    ptr = rx_queue.begin();
      }
      // remove it
      if ( ptr != rx_queue.end() )
      {
         rx_queue.erase(ptr);
         rx_changed = true;
         dropped_msg_count++;
      }
   }
   
   // add new message
   // prio: add to end of prio messages
   if ( msg.get_class() == s3_message::PRIORITY )
   {
      T_RX_Queue::iterator ptr;
      // find first non-prio - just after newest prio
      for (ptr = rx_queue.begin(); 
      ptr != rx_queue.end();
      ptr++)
      {
    if ((*ptr).get_class() == s3_message::NORMAL) break;
      }
      // no non-prio: insert new at end
      if ( ptr == rx_queue.end() ) {
    rx_queue.push_back(msg);
    rx_changed = true;
      }
      // add before start of non-prios (i.e. at end of prios)
      else
      {
    rx_queue.insert(ptr, msg);
    rx_changed = true;
      }
   }
   // non-prio: add to end
   else
   {
      rx_queue.push_back(msg);
      rx_changed = true;
   }
   // notify everyone in arrival_notification_list
   if ( rx_changed )
   {
      for ( std::list<s3_semaphore*>::const_iterator ptr = 
          arrival_notification_list.begin();
       ptr != arrival_notification_list.end();
       ptr++ )
      {
    (*ptr)->post();
      }
   }
   state_lock.unlock();
}

// format error string - embed name
std::string s3_message_box::err_str(const std::string& err)
{
   std::stringstream str;
   str << "[name = '" << name << "']: "
       << err;
   return str.str();
}

//
std::string s3_message_box::get_next_message_id()
{
   std::stringstream str;
   str << name << "/" << seq_nr;
   seq_nr++;
   return str.str();
}

//
s3_post_office::s3_post_office() :
   conn(0),
   ctrl_rx_queue(10),
   client_rx_queue(10),
   seq_nr(0)
{
}

// destructor
s3_post_office::~s3_post_office()
{
   if ( is_connected() )
   {
      disconnect();
   }   
}

#if S3_HAS_LOCAL_PO_SWITCH
//
void s3_post_office::connect( void )
{
   if (conn)
   {
      throw s3_generic_exception("s3_post_office::connect()",
             "Already connected");
   }
   conn = new s3_txport_local<s3_message>;

   if ( ! s3_post_office_switch::po_switch_txp_ptr  )
   {
      throw s3_generic_exception("s3_post_office::connect()",
                                 "post office switch not running");

   }
   conn_id = conn->connect( s3_post_office_switch::po_switch_txp_ptr );
   
   if (! conn->is_connected_by_id(conn_id))
   {
      // we may go out of scope here, so we should delete conn.
      delete conn;
      conn = 0;
      throw s3_generic_exception("s3_post_office::connect()",
             "Connection failed");
   } 

   // start myself
   start();
   // wait for actual start before returning
   started.wait();
}
#else
//
void s3_post_office::connect(const std::string& ip,
              int port)
{
   if (conn)
   {
      throw s3_generic_exception("s3_post_office::connect()",
             "Already connected");
   }
   conn = new s3_txport_tcp<s3_message>;
   conn_id = conn->connect(ip, port);

   if (! conn->is_connected_by_id(conn_id))
   {
      // we may go out of scope here, so we should delete conn.
      delete conn;
      conn = 0;
      throw s3_generic_exception("s3_post_office::connect()",
             "Connection failed");
   }
   // start myself
   start();
   // wait for actual start before returning
   started.wait();
}
#endif


//
void s3_post_office::disconnect()
{
   // just do nothing if we're not connected (not a sin)
   if ( ! conn ) return;
   // throw if there are still connected clients.
   if ( ! dst_map.empty() )
   {
      throw s3_generic_exception("s3_post_office::disconnect()",
             "Connected clients present");
   }
   // stop event loop, disconnect and deallocate
   try
   {
      conn->stop();
      conn->wait_on_exit();
      delete conn;
      conn = 0;      
   }
   catch(const std::exception& e)
   {
      std::cerr << "s3_post_office::disconnect() " << e.what()
      << std::endl << std::flush;
   }
}

//
void s3_post_office::add_dst(const std::string& dst,
              s3_message_box& mbox,
              bool is_group)
{
   // atomic register and add operation
   dst_lock.lock();
   // if no connection, throw
   if ( ! is_connected() ) 
   {
      dst_lock.unlock();
      throw s3_generic_exception("s3_post_office::add_dst()",
             "No connection");
   }
   bool present = (dst_map.count(dst) == 1);
   // if destination is not a group and already in map, throw  [/g p]  
   if ((! is_group) && present)
   {
      dst_lock.unlock();
      throw s3_generic_exception("s3_post_office::add_dst()",
             "Name clash (local)");
   }
   // if destination is not present in map register with switch, check for
   // uniqueness and add to map  [ g /p || /g /p = /p ]
   if (! present)
   {
      // this should be able to proceed *with* a state lock in place
      if ( add_dst_switch(dst, is_group) ) 
      {
    // atomic map update
    lock_state();
    dst_map[dst] = dst_data(std::list<s3_message_box*>(1, &mbox), 
             is_group);
    unlock_state();
      }
      else
      {
    dst_lock.unlock();
    throw s3_generic_exception("s3_post_office::add_dst()",
                "Name clash");
      }
   }
   // group and in map - uniqueness already verified and already registered
   // with switch [g p]
   else
   {
      // atomic map update
      lock_state();
      dst_map[dst].clients.push_back(&mbox);
      unlock_state();
   }
   dst_lock.unlock();
}

//
void s3_post_office::remove_dst(const std::string& dst,
            s3_message_box& mbox,
            bool is_group )
{
   // atomic deregister and remove operation
   dst_lock.lock();
   // if no connection, throw
   if ( ! is_connected() ) 
   {
      dst_lock.unlock();
      throw s3_generic_exception("s3_post_office::remove_dst()",
             "No connection");
   }
   bool present = (dst_map.count(dst) == 1);
   // if destination is not in map, throw
   if (! present)
   {
      dst_lock.unlock();
      throw s3_generic_exception("s3_post_office::remove_dst()",
             "Destination not registered");
   }
   // if only single client in list unregister with switch and remove entry
   // from map
   if ( dst_map[dst].clients.size() == 1 )
   {
      remove_dst_switch(dst, is_group);
      // atomic map update
      lock_state();
      dst_map.erase(dst);
      unlock_state();
   }
   // remove client from list
   else
   {
      // atomic map update
      lock_state();
      std::list<s3_message_box*>::iterator c =
    std::find(dst_map[dst].clients.begin(),
         dst_map[dst].clients.end(),
         &mbox);
      dst_map[dst].clients.erase(c);
      unlock_state();
   }
   dst_lock.unlock();
}

//
void s3_post_office::main_loop()
{
   if (! is_connected() )
   {
      throw s3_generic_exception("s3_post_office::main_loop()",
             "No connection");
   }

   // incoming queue from post office switch
   s3_fifo_queue<s3_txport_data<s3_message> >& switch_rx_queue =
      conn->get_rx_queue();
   s3_fifo_queue<s3_txport_data<s3_message> >& switch_tx_queue =
      conn->get_tx_queue();  

   // semaphore used to synch to queue entries. subscribe this to
   // switch_rx_queue and client_rx_queue as consumers
   s3_semaphore rx_rdy;

   switch_rx_queue.subscribe_consumer(rx_rdy);
   client_rx_queue.subscribe_consumer(rx_rdy);
  
   /*
    * This periodic notifier is used to dislodge the rx_rdy semaphore.
    * The semaphore is posted when switch_rx_queue and client_rx_queue
    * received data.
    *
    * Without the notifier, rx_rdy.wait() will block the thread. This
    * poses a problem in the WIN32 port, where phtread_cancel does not
    * work correctly.
    *
    * THe periodic notifier is not required on the Unix platform,
    * but it provides a cleaner termination strategy.
    *
    * The periodic notifier's timeout determines what the maximum time
    * is that you will wait after request_terminate() has been
    * called. Feel free to experiment.
    * 
    * This notifier is a requirement for the WIN32 port.
    * 30 May 2002: Jan Pool <jpool@stonethree.com>
    */
   s3_periodic_notifier term_pn(1000);
   term_pn.subscribe(rx_rdy);
   term_pn.enable();

   // message and dispatcher
   //s3_message msg;
   //s3_message_box* org = 0;
   client_msg cm;
   // use for alternating between input queues
   bool twiddle = true;

   // signal all is go and spin
   started.post();
   // Use test_terminate to protect is_connected. [jpool]
   while ( !test_terminate() && is_connected() ) 
   {
      // Modified the while to make the termination easier. [jpool]

      int oper = 0;

      // wait for queue activity
      rx_rdy.wait();      
      
      /*
       * We must test for termination here, since some of the attributes
       * do not exist anymore. We must therefore break out before trying
       * to access any of them.
       * [jpool]
       */
      if ( test_terminate() )
      {
    break;   
      } // if test_terminate()       
       
      if ( twiddle )
      {
    if ( ! switch_rx_queue.empty() ) 
    {
       oper = 1;
       twiddle = false;
    }
    else if ( ! client_rx_queue.empty() ) 
    {
       oper = 2;
       twiddle = true;
    }
      }
      else
      {
    if ( ! client_rx_queue.empty() ) 
    {
       oper = 2;
       twiddle = true;
    }
    else if ( ! switch_rx_queue.empty() ) 
    {
       oper = 1;
       twiddle = false;
    } 
      }      
      
      // If we received data, we process it here.
      if ( oper != 0)
      {
    // Got data.
    if ( oper == 1 ) // read switch
    {
       cm.msg = switch_rx_queue.pop().data;
       cm.client = 0;
       S3FC_DBG( std::cerr << "s3_post_office::main_loop() [" << this
            << "] " << "reading from switch queue" << std::endl
            << std::flush);
    } // oper == 1
    else if ( oper == 2 )  // read client
    {
       cm = client_rx_queue.pop();
       S3FC_DBG( std::cerr << "s3_post_office::main_loop() [" << this
            << "] " << "reading from " << "client queue"
            << std::endl << std::flush);
    } // oper = 2
    
    // cm now effectively contains the next message removed from
    // rx_queue, with its originator
    S3FC_DBG( std::cerr << "s3_post_office::main_loop() [" << this
         << "] " << "message received (" << cm.client << "): "
         << std::endl << "  " << cm.msg.get_debug_str()
         << std::endl << std::flush );
    
    // message routed into control queue
    if (cm.msg.get_class() == s3_message::CONTROL)
    {
       S3FC_DBG( std::cerr << "s3_post_office::main_loop() [" << this
            << "] " << "routing msg into ctrl_rx_queue"
            << std::endl << std::flush );
       ctrl_rx_queue.push(cm.msg);
    }
    // regular message
    else 
    {
            // protect the map while reading it
            lock_state();

       const bool present = dst_map.count( cm.msg.get_to() ) > 0;
            const bool group = present ?
               dst_map[cm.msg.get_to()].group : false;

            const std::list<s3_message_box*>& clients = present ?
               dst_map[cm.msg.get_to()].clients : std::list<s3_message_box*>();
       
            // origin is switch: forward to all in clients
       if ( cm.client == 0 )
       {
          deliver_all_but_one(cm.msg, clients, 0);
       } // if cm.client
       // origin is client
       else
       {
          // dest not present: forward to switch
          if ( ! present )
          {
        S3FC_DBG( std::cerr << "s3_post_office::main_loop() ["
             << this << "] " << "forwarding msg to (switch)"
             << std::endl << std::flush );
        switch_tx_queue.push(s3_txport_data<s3_message>(cm.msg,
                          conn_id));
          }
          // dest is present and not group: forward to single list entry
          else if ( ! group )
          {
        // sanity check
        if ( clients.size() != 1 )
        {
           throw s3_generic_exception("s3_post_office::main_loop()",
                  "Internal error: "
                  "Non-group dest with "
                  "clients.size()> 1 ");
           
        }
        S3FC_DBG( std::cerr << "s3_post_office::main_loop() ["
             << this << "] " << "forwarding msg to (client "
             << clients.front() << ")" << std::endl
             << std::flush );
        clients.front()->deliver_msg(cm.msg);
          }
          // dest is present and group: forward to clients in list except
          //   originating AND forward to switch
          else
          {
        deliver_all_but_one(cm.msg, clients, cm.client);
        S3FC_DBG( std::cerr << "s3_post_office::main_loop() ["
             << this << "] forwarding msg to " << "(switch)"
             << std::endl << std::flush );
        switch_tx_queue.push(
           s3_txport_data<s3_message>(cm.msg, conn_id));
          }
       }
            unlock_state();
         //   dst_lock.unlock();
    } 
      }
   }
   
   // cleanup periodic notifier stuff. [jpool]
   term_pn.disable();
   term_pn.unsubscribe(rx_rdy);

   return;
}


//
void s3_post_office::dispatch_msg(const s3_message& msg,
              s3_message_box& mbox)
{
   client_rx_queue.push(client_msg(msg, &mbox));
}

//
bool s3_post_office::is_connected() const
{
   // we love short-circuiting boolean expressions
   return (conn && conn->is_connected_by_id(conn_id));
}

//
bool s3_post_office::add_dst_switch(const std::string& name,
                bool is_group)
{

   std::stringstream body;
   body << "add_dst" 
   << " " << (is_group ? "group" : "individual") 
   << " " << name;
    
   // create control messages requesting add_dst and dispatch
   s3_message req("__post_office__", 
        "__post_office_switch__", 
        body.str(),
        s3_message::CONTROL,
        get_next_message_id());
   S3FC_DBG( std::cerr
        << "s3_post_office::add_dst_switch() Sending control message: "
        << std::endl << "  "
        << req.get_debug_str() << std::endl << std::flush);
   conn->get_tx_queue().push(s3_txport_data<s3_message>(req, conn_id));

   // wait for response
   s3_message resp;
   resp = ctrl_rx_queue.pop();

   return (resp.get_body() == "success");
}

//
bool s3_post_office::remove_dst_switch(const std::string& name,
                   bool is_group)
{
   std::stringstream body;
   body << "remove_dst" 
   << " " << (is_group ? "group" : "individual") 
   << " " << name;
    
   // create control messages requesting add_dst and dispatch
   s3_message req("__post_office__", 
        "__post_office_switch__", 
        body.str(),
        s3_message::CONTROL,
        get_next_message_id());
   S3FC_DBG( std::cerr << "s3_post_office::remove_dst_switch() "
        << "Sending control message: " << std::endl << "  "
        << req.get_debug_str() << std::endl << std::flush);
   conn->get_tx_queue().push(s3_txport_data<s3_message>(req, conn_id));

   // wait for response
   s3_message resp;
   resp = ctrl_rx_queue.pop();

   return (resp.get_body() == "success");
}

//
void s3_post_office::deliver_all_but_one(
   const s3_message& msg,
   const std::list<s3_message_box*>& clients,
   const s3_message_box* except)
{
   for (std::list<s3_message_box*>::const_iterator ptr = clients.begin();
   ptr != clients.end();
   ptr++)
   {
      if ( *ptr != except ) 
      {
    S3FC_DBG( std::cerr << "s3_post_office::main_loop() "
         << "forwarding msg to (client "
         << *ptr << ")" << std::endl << std::flush );

    (*ptr)->deliver_msg(msg);
      }
   }
}

//
std::string s3_post_office::get_next_message_id()
{
   std::stringstream str;
   str << "__post_office__" << "/" << seq_nr;
   seq_nr++;
   return str.str();
}

#if S3_HAS_LOCAL_PO_SWITCH
s3_txport_local<s3_message>* s3_post_office_switch::po_switch_txp_ptr = 0;
#endif

//
s3_post_office_switch::txp_event_queuer::~txp_event_queuer()
{
}

//
void s3_post_office_switch::txp_event_queuer::operator()(
   const s3_txport_event& event)
{
   // this can (and should) never block
   ev_queue.push(event);
}

//
s3_post_office_switch::dst_data::dst_data()
{
}

//
s3_post_office_switch::dst_data::dst_data(
   const std::list<client_id_t>& n_clients,
   bool n_group) :
   clients(n_clients),
   group(n_group)
{
}

//
s3_post_office_switch::s3_post_office_switch(int n_port) :
   conn(0),
   seq_nr(0)
{
#if S3_HAS_LOCAL_PO_SWITCH
   conn = new s3_txport_local<s3_message>;
   po_switch_txp_ptr = conn;
#else
   conn = new s3_txport_tcp<s3_message>;
   conn->listen(n_port);

 
   if ( ! conn->listening() )
   {
      // check status - throw if problem
      delete conn;
      conn = 0;
      throw s3_generic_exception("s3_post_office_switch"
             "::s3_post_office_switch()",
             "Init failed, port may be in use.");
   }
#endif

   // attach event queuer to transport
   conn->subscribe_handler(txp_event_queuer);
   // start myself
   start();
   // wait for start to be indicated
   started.wait();
}

//
s3_post_office_switch::~s3_post_office_switch()
{
   if (is_running())
   {
      // Use soft terminate. [jpool]
      request_terminate();
      wait_on_exit();
   }
   conn->stop();
}

//
void s3_post_office_switch::main_loop()
{
   // queue through which we receive messages
   queue_t& rx_queue = conn->get_rx_queue();
   // this should get posted if any of the rx queues (message or transport
   // event) receive something
   s3_semaphore rdy;
   rx_queue.subscribe_consumer(rdy);
   txp_event_queuer.ev_queue.subscribe_consumer(rdy);
   
   /*
    * This periodic notifier is used to dislodge the rdy semaphore.
    * The semaphore is posted when rx_queue and txp_event_queuer
    * received data.
    *
    * Without the notifier, rdy.wait() will block the thread. This
    * poses a problem in the WIN32 port, where phtread_cancel does not
    * work correctly.
    *
    * THe periodic notifier is not required on the Unix platform,
    * but it provides a cleaner termination strategy.
    *
    * The periodic notifier's timeout determines what the maximum time
    * is that you will wait after request_terminate() has been
    * called. Feel free to experiment.
    * 
    * This notifier is a requirement for the WIN32 port.
    * 30 May 2002: Jan Pool <jpool@stonethree.com>
    */   
   s3_periodic_notifier term_pn(1000);
   term_pn.subscribe(rdy);
   term_pn.enable();

   // indicate that we have started
   started.post();
   
   // run until the transport dies or we get stopped
   while ( conn->listening() )
   {
      rdy.wait();

      /*
       * We must test for termination here, since some of the attributes
       * do not exist anymore. We must therefore break out before trying
       * to access any of them.
       * [jpool]
       */
      if ( test_terminate() )  
      {
    break;
      } // if test_terminate() 

      // service all pending transport events
      if (! txp_event_queuer.ev_queue.empty())
      {
    handle_transport_events();
      }
      // service the rx queue
      if (! rx_queue.empty())
      {
    // we know there is something in the queue - never block
    s3_txport_data<s3_message> packet = rx_queue.pop();
    S3FC_DBG( std::cerr << "s3_post_office_switch::main_loop() "
         << "recvd message from (" << packet.get_id() << "): "
         << std::endl << "  " << packet.data.get_debug_str()
         << std::endl << std::flush );
    
    // control message ?
    if ( packet.data.get_class() == s3_message::CONTROL) 
    {
       std::vector<std::string> cmd = 
          s3_conversion::string_to_vec_string(packet.data.get_body());
       // 
       if ( cmd.size() != 3 )
       {
          std::stringstream lmsg;
          lmsg << "Invalid command length: size = " << cmd.size();
          for ( unsigned int i = 0; i < cmd.size(); i++ )
          {
        lmsg << " cmd[" << i << "]: '" << cmd[i] << "'";
          }
          lmsg << "Message dump: " << packet.data.get_debug_str()
          << std::endl;
          log_msg("main_loop()", lmsg.str());
       }
       bool success = false;
       
       bool is_group = (cmd[1] == "group");
       std::string dst = cmd[2];
       client_id_t id = packet.get_id();
       if ( cmd[0] == "add_dst" )
       {
          success = add_dst(dst, id, is_group);
          S3FC_DBG( std::cerr << "s3_post_office_switch::main_loop()" 
          << " add_dst said: " << success
          << std::endl << std::flush );
       }
       else if ( cmd[0] == "remove_dst" )
       {
          success = remove_dst(dst, id, is_group);
          S3FC_DBG( std::cerr << "s3_post_office_switch::main_loop()" 
          << " remove_dst said: " << success
          << std::endl << std::flush );
       }
       else 
       {
          success = false;
          std::stringstream lmsg;
          lmsg << "main_loop()" 
          << " unknown command: '" 
          << cmd[1] << "'";
          log_msg("main_loop",
             lmsg.str());
       }
       
       if ( success ) 
       {
          reply_success(packet);
       }
       else
       {
          reply_failure(packet);
       }
    }
    // normal packets are routed between client connections
    else 
    {
       // where it came from
       client_id_t src_client = packet.get_id();
       // where it should go
       std::list<client_id_t> dst_clients;
       bool present = false;
       //
       lock_state();
       if ( dst_map.count(packet.data.get_to()) )
       {
          dst_clients = dst_map[packet.data.get_to()].clients;
          present = true;
       }
       unlock_state();
       // only forward if destination present in map
       if ( present ) 
       {
          S3FC_DBG( std::cerr << "s3_post_office_switch::main_loop() " 
          << dst_clients.size() << " clients in list"
          << std::endl << std::flush );
          lock_state();
          for (std::list<client_id_t>::const_iterator ptr =
             dst_clients.begin();
          ptr != dst_clients.end();
          ptr++)
          {
        // don't forward to originating client
        if ( *ptr != src_client )
        {
           S3FC_DBG( std::cerr
                << "s3_post_office_switch::main_loop()" 
                << " forwarding to (" << packet.get_id()
                << ")" << std::endl << std::flush );
           forward_msg(packet.data, *ptr);
        }
          }
          unlock_state();
       }
       else
       {
          S3FC_DBG( std::cerr << "s3_post_office_switch::main_loop()" 
          << " dropped message - no clients "
          << std::endl << std::flush );
       }
    }
      }
   }

   // Clean up periodic notifier stuff. [jpool]
   term_pn.unsubscribe(rdy);
   term_pn.disable();

   return;
}

// associate client with dst. check for ambiguous names (if not 
// a group).
// 
bool s3_post_office_switch::add_dst(const std::string& dst,
                client_id_t client,
                bool is_group)
{
   bool success = false;
   bool present = (dst_map.count(dst) == 1);

   // if destination is a group OR an individual that is NOT present,
   // we may proceed - else someone already added individual with 
   // existing name and we should fail
   if ( is_group || ! present )
   {
      // dst already there - client should be added to list if its
      // not already there
      if ( present )
      {
    std::list<client_id_t>& c_clients = dst_map[dst].clients;
    // only add if client not already in list
    if ( std::find(c_clients.begin(), c_clients.end(), client) ==
         c_clients.end() )
    {
       c_clients.push_back(client);
    }
      }
      // create entry for new destination in the map and add client as
      // the only one in the client list
      else
      {

    dst_map[dst] = dst_data(std::list<client_id_t>(1, client),
             is_group);
      }
      success = true;
   }
   // log failure
   if ( ! success )
   {
      std::stringstream lmsg;
      lmsg << "Failure: "
      << "[dst = '" << dst << "'"
      << ", client = '" << client << "'"
      << ", is_group = '" << "'"
      << ", present = '" << present << "'"
      << "]";
      log_msg("add_dst()", lmsg.str());
   }

   return success;
}

// remove an association between dst and client. if the client is the
// only one in the client list, the destination is entirely removed from
// the map. 
// a check is first performed to see whether the association actually 
// exists and has the supplied type - if this fails, then no action is 
// taken, but a true is returned
bool s3_post_office_switch::remove_dst(const std::string& dst,
                   client_id_t client,
                   bool is_group)
{
   bool success = false;
   bool present = (dst_map.count(dst) == 1);
   
   // only attempt remove if present
   if (present)
   {
      // check whether registered name coresponds to passed arguments,
      // i.e. in map and with correct type (indiv/group)
      if(dst_map[dst].group == is_group)
      {
    // iterator referencing client associated with dst
    std::list<client_id_t>::iterator cp =
       std::find(dst_map[dst].clients.begin(),
            dst_map[dst].clients.end(),
            client);
    if ( cp != dst_map[dst].clients.end() )
    {
       dst_map[dst].clients.erase(cp);
       // if dst entry in dst_map has no more clients in list, it should
       // be removed from the map
       if ( dst_map[dst].clients.size() == 0 )
       {
          dst_map.erase(dst);
       }
       success = true;
    }
      }
   }
   // 
   else
   {
      success = true;
   }
   // log failure
   if ( ! success )
   {
      std::stringstream lmsg;
      lmsg << "Failure: "
      << "[dst = '" << dst << "'"
      << ", client = '" << client << "'"
      << ", is_group = '" << "'"
      << ", present = '" << present << "'"
      << ", dst_map[dst].group = '" << dst_map[dst].group << "'"
      << "]";
      log_msg("remove_dst()", lmsg.str());
   }
   return success;
}

//
void s3_post_office_switch::remove_client(client_id_t client)
{
   // iterate through map, checking for the presence of client in each
   // entry's client list and then removing it. if this results in the
   // list being empty - push associated destination key into the prune list
   std::vector<std::string> prune_list;
   for (std::map<std::string, dst_data>::iterator map_iter = dst_map.begin();
   map_iter != dst_map.end();
   ++map_iter)
   {
      // remove elements with client in client_list corresponding to map
      // entry at map_iter
      map_iter->second.clients.remove(client);
      // add to prune list if empty
      if (map_iter->second.clients.size() == 0)
      {
    prune_list.push_back(map_iter->first);
      }
   }
   // remove destinations with empty client lists
   for (std::vector<std::string>::iterator iter = prune_list.begin();
   iter != prune_list.end();
   ++iter)
   {
      dst_map.erase(*iter);
   }
}

// 
void 
s3_post_office_switch::reply_success(const s3_txport_data<s3_message>& packet)
{
   const s3_message& msg = packet.data;
   s3_message reply_msg("__post_office_switch__",
         msg,
         "success",
         get_next_message_id());
   forward_msg(reply_msg, packet.get_id());
}

// 
void 
s3_post_office_switch::reply_failure(const s3_txport_data<s3_message>& packet)
{
   const s3_message& msg = packet.data;
   s3_message reply_msg("__post_office_switch__",
         msg,
         "failure",
         get_next_message_id());
   forward_msg(msg, packet.get_id());
}

//
void s3_post_office_switch::forward_msg(const s3_message& msg,
               const client_id_t client)
{
   S3FC_DBG( std::cerr << "s3_post_office_switch::forward_msg() "
        << "forwarding to (" << client << "): "
        << std::endl << "  "
        << msg.get_debug_str() << std::endl << std::flush );
   conn->get_tx_queue().push(s3_txport_data<s3_message>(msg, client));
}

// 
void s3_post_office_switch::log_msg(const std::string& where,
                const std::string& what)
{
   std::cerr << "s3_post_office_switch::" << where << " [LOGMSG] [" 
        << what << "]" << std::endl << std::flush;
}

//
std::string s3_post_office_switch::get_next_message_id()
{
   std::stringstream str;
   str << "__post_office_switch__" << "/" << seq_nr;
   seq_nr++;
   return str.str();
}

void s3_post_office_switch::handle_transport_events()
{
   // handle all pending events
   while (! txp_event_queuer.ev_queue.empty())
   {
      s3_txport_event ev = txp_event_queuer.ev_queue.pop();
      // only handle disconnect events - ignore the rest
      if (ev.event == s3_txport_event::disconnect)
      {
    remove_client(ev.id);
      }
   }
}

---

Send comments to: s3fc@stonethree.com