S3FC project page

S3FC home page

---

               

---

s3_message.h

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.
 */

#ifndef S3_MESSAGE_H
#define S3_MESSAGE_H

/*
 * - The message is distributed to all client 
 * Currently, messages are routed according to an @e extremely simple (and 
 * admittedly ineffective, inelegant and really stupid) algorithm:
 * - Any received message is forwarded to all connections in the connection list 
 *   of that destination, except the one on which it was received.
 */

//#define S3_HAS_LOCAL_PO_SWITCH 1

#include <iostream>
#include <string>
#include <vector>
#include <list>
#include <s3fc/s3_streamable.h>
#include <s3fc/s3_fifo_queue.h>
#include <s3fc/s3_growable_fifo_queue.h>
#include <s3fc/s3_txport_base.h>

#if S3_HAS_LOCAL_PO_SWITCH
   #include <s3fc/s3_txport_local.h>

#else
   #include <s3fc/s3_txport_tcp.h>
#endif   

#include <s3fc/s3_mutex.h>
#include <s3fc/s3_semaphore.h>

class s3_message_box;

class s3_message : public s3_streamable
{
  public:
   typedef std::string msg_id_t;
   enum msg_class_t
   {
      NORMAL,
      PRIORITY,
      CONTROL
   };
  protected:
   bool empty;
   std::string from;
   std::string to;
   std::string body;
   msg_class_t mc;
   msg_id_t id;
   msg_id_t reply_id;
  public:
   S3_DECLARE_MAGIC_STRING;
   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);
   s3_message(const std::string& n_from,
         const s3_message& msg,
         const std::string& n_body,
         const msg_id_t& n_id);
   virtual ~s3_message();
   virtual void pack(s3_pack_buffer &b) const;
   virtual void unpack(const s3_pack_buffer &b);
   bool is_empty() const;
   bool is_reply(const msg_id_t& n_id) const;
   std::string get_from() const;
   std::string get_to() const;
   std::string get_body() const;
   msg_class_t get_class() const;
   // diagnostic purposes - return a formatted string containing
   // all the field values
   std::string get_debug_str() const;
};

S3_TYPEGROUP_TRAITS_POD(s3_message::msg_class_t);

// forward decl
class s3_post_office;

class s3_message_box
{
   // provide access to privates 
   friend class s3_post_office;
  public:
   typedef std::list<s3_message> T_RX_Queue;

   static unsigned int dropped_msg_count;
  private:
   static s3_post_office* poffice;
   bool connected;
   std::string name;
   unsigned int rx_queue_len;
   T_RX_Queue rx_queue;
   std::list<std::string> subscription_list;
   int seq_nr;
   std::list<s3_semaphore*> arrival_notification_list;
   s3_mutex state_lock;
  public:
   s3_message_box(const std::string& name = "",
        unsigned int n_queue_len = 20);
   ~s3_message_box();
   static void init(const std::string& ip_addr,
          int port);
   static void de_init();
   static bool initialised();
   void set_name(const std::string& name);
   void connect();
   void disconnect();
   void subscribe_group(const std::string& group);
   void unsubscribe_group(const std::string& group);
   void subscribe_arrival(const s3_semaphore& sem);
   s3_message::msg_id_t send_msg(const std::string& dst,
             const std::string& body,
             bool priority = false);
   s3_message::msg_id_t reply_msg(const s3_message& msg,
              const std::string& body = "");
   s3_message get_msg();
  protected:
   void deliver_msg(const s3_message& msg);
   std::string err_str(const std::string& err);
   std::string get_next_message_id();
   
};

class s3_post_office : public s3_thread_base
{
   friend class s3_message_box;
  private:
   struct client_msg
   {
      // message
      s3_message msg;
      // client that dispatched it
      s3_message_box* client;
      // 
      client_msg() { }
      // 
      client_msg(const s3_message& n_msg,
       s3_message_box* n_client) :
    msg(n_msg),
    client(n_client) { }
   };
   struct dst_data
   {
      std::list<s3_message_box*> clients;
      bool group;
      dst_data()  { }
      dst_data(const std::list<s3_message_box*> n_clients,
          bool n_group) :
    clients(n_clients),
    group(n_group)   {  }
   };
   
#if S3_HAS_LOCAL_PO_SWITCH
   s3_txport_local<s3_message>* conn;
#else
   s3_txport_tcp<s3_message>* conn;
#endif

   s3_txport_client_id conn_id;
   std::map<std::string, dst_data> dst_map;
   s3_fifo_queue<s3_message> ctrl_rx_queue;
   s3_fifo_queue<client_msg> client_rx_queue;
   int seq_nr;
   s3_mutex dst_lock;
   s3_semaphore started;
  public:
   
   ~s3_post_office();
   
   s3_post_office();
#if S3_HAS_LOCAL_PO_SWITCH
   void connect( void );
#else
   void connect(const std::string& ip, int port);
#endif

   void disconnect();
   void add_dst(const std::string& dst,
      s3_message_box& mbox,
      bool is_group);
   void remove_dst(const std::string& dst,
         s3_message_box& mbox,
         bool is_group);
   virtual void main_loop();
  protected:
   void dispatch_msg(const s3_message& msg,
           s3_message_box& mbox);
   bool is_connected() const;
   bool add_dst_switch(const std::string& name,
             bool is_group);
   bool remove_dst_switch(const std::string& name,
           bool is_group);
   void deliver_all_but_one(const s3_message& msg,
             const std::list<s3_message_box*>& clients,
             const s3_message_box* except);
   std::string get_next_message_id();
};

class s3_post_office_switch : public s3_thread_base
{
  private:
   struct txp_event_queuer : public s3_event_handler<s3_txport_event>
   {
     public:
      virtual ~txp_event_queuer();
      typedef s3_growable_fifo_queue<s3_txport_event> ev_queue_t;
      // event queue: 
      ev_queue_t ev_queue;
      virtual void operator()(const s3_txport_event& event);
   };
   typedef s3_txport_client_id client_id_t;
   // forward decl before friend and friend before proper decl
   struct dst_data;
   friend struct dst_data;
   struct dst_data
   {
      std::list<client_id_t> clients;
      bool group;
      // Default constructor - empty
      dst_data();
      // Initialising constructor.
      dst_data(const std::list<client_id_t>& n_clients,
          bool n_group);
   };
   typedef s3_fifo_queue<s3_txport_data<s3_message> > queue_t;
#if S3_HAS_LOCAL_PO_SWITCH
   s3_txport_local<s3_message>* conn;
#else
   s3_txport_tcp<s3_message>* conn;
#endif

   txp_event_queuer txp_event_queuer;
   std::map<std::string, dst_data> dst_map;
   int seq_nr;
   s3_semaphore started;
  public:
#if S3_HAS_LOCAL_PO_SWITCH
     static s3_txport_local<s3_message>* po_switch_txp_ptr;
#endif

   s3_post_office_switch(int n_port);
   // now 20:15 - will do docs later. this does stop() and wait_on_exit()
   virtual ~s3_post_office_switch();
   virtual void main_loop();
  protected:

   bool add_dst(const std::string& dst,
      client_id_t client,
      bool is_group);
   bool remove_dst(const std::string& dst,
         client_id_t client,
         bool is_group);
   void remove_client(client_id_t client);
   void reply_success(const s3_txport_data<s3_message>& packet);
   
   void reply_failure(const s3_txport_data<s3_message>& packet);
   void forward_msg(const s3_message& msg,
          const client_id_t client);
   void log_msg(const std::string& where,
      const std::string& what);
   std::string get_next_message_id();
   void handle_transport_events();
};

#endif

---

Send comments to: s3fc@stonethree.com