ACE/apps/Gateway/Peer/Peer.h

   1 /* -*- C++ -*- */
   2
   3 //=============================================================================
   4 /**
   5  *  @file    Peer.h
   6  *
   7  *  These classes process Supplier/Consumer events sent from the
   8  *  gateway (gatewayd) to its various peers (peerd).  The general
   9  *  collaboration works as follows:
  10  *
  11  *  1. <Peer_Acceptor> creates a listener endpoint and waits
  12  *     passively for gatewayd to connect with it.
  13  *
  14  *  2. When a gatewayd connects, <Peer_Acceptor> creates an
  15  *     <Peer_Handler> object that sends/receives events from
  16  *     gatewayd on that connection.
  17  *
  18  *  3. The <Peer_Handler> waits for gatewayd to inform it of its
  19  *     connection ID, which is prepended to all subsequent outgoing
  20  *     events sent from peerd.
  21  *
  22  *  4. Once the connection ID is set, peerd periodically sends events
  23  *     to gatewayd.  Peerd also receives and "processes" events
  24  *     forwarded to it from gatewayd.  In this program, peerd
  25  *     "processes" the events sent to it by writing them to stdout.
  26  *
  27  *  Note that in the current peerd implementation, one Peer process
  28  *  cannot serve as both a Consumer and Supplier of Events.  This is
  29  *  because the gatewayd establishes a separate connection for
  30  *  Suppliers and Consumers and the peerd only maintains a single
  31  *  <Peer_Handler> object to handle this one connection.  Enhancing
  32  *  this implementation to be both a Consumer and Supplier
  33  *  simultaneously is straightforward, however.  In addition,
  34  *  multiple peerd processes can already work together to play these
  35  *  different roles.
  36  *
  37  *  @author Douglas C. Schmidt
  38  */
  39 //=============================================================================
  40
  41
  42 #ifndef PEER_H
  43 #define PEER_H
  44
  45 #include "ace/Service_Config.h"
  46
  47 #if !defined (ACE_LACKS_PRAGMA_ONCE)
  48 # pragma once
  49 #endif /* ACE_LACKS_PRAGMA_ONCE */
  50
  51 #include "ace/Acceptor.h"
  52 #include "ace/SOCK_Acceptor.h"
  53 #include "ace/SOCK_Connector.h"
  54 #include "ace/Svc_Handler.h"
  55 #include "ace/Connector.h"
  56 #include "ace/Null_Condition.h"
  57 #include "ace/Null_Mutex.h"
  58 #include "Options.h"
  59
  60 ACE_SVC_FACTORY_DECLARE (Peer_Factory)
  61
  62 #if defined ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION_EXPORT
  63 template class ACE_Svc_Export ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>;
  64 #endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION_EXPORT */
  65
  66 /**
  67  * @class Peer_Handler
  68  *
  69  * @brief Handle Peer events arriving from a Gateway.
  70  */
  71 class ACE_Svc_Export Peer_Handler : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>
  72 {
  73 public:
  74   /// Initialize the peer.
  75   Peer_Handler ();
  76
  77   /// Shutdown the Peer.
  78   ~Peer_Handler ();
  79
  80   /// Initialize the handler when called by
  81   /// <ACE_Acceptor::handle_input>.
  82   virtual int open (void * = 0);
  83
  84   /// Receive and process peer events.
  85   virtual int handle_input (ACE_HANDLE);
  86
  87   /// Send a event to a gateway (may be queued if necessary due to flow
  88   /// control).
  89   virtual int put (ACE_Message_Block *, ACE_Time_Value *tv = 0);
  90
  91   /// Finish sending a event when flow control conditions abate.
  92   virtual int handle_output (ACE_HANDLE);
  93
  94   /// Periodically send events via <ACE_Reactor> timer mechanism.
  95   virtual int handle_timeout (const ACE_Time_Value &,
  96                               const void *arg);
  97
  98   /// Perform object termination.
  99   virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE,
 100                             ACE_Reactor_Mask = ACE_Event_Handler::ALL_EVENTS_MASK);
 101
 102 protected:
 103   typedef ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH> inherited;
 104
 105   /// Transmit <mb> to the gatewayd.
 106   int transmit (ACE_Message_Block *mb,
 107                 size_t n,
 108                 int event_type);
 109
 110   /// Receive an Peer event from gatewayd.
 111   virtual int recv (ACE_Message_Block *&mb);
 112
 113   /// Send an Peer event to gatewayd, using <nonblk_put>.
 114   virtual int send (ACE_Message_Block *mb);
 115
 116   /// Perform a non-blocking <put>, which tries to send an event to the
 117   /// gatewayd, but only if it isn't flow controlled.
 118   virtual int nonblk_put (ACE_Message_Block *mb);
 119
 120   /// Register Consumer subscriptions with the gateway.
 121   int subscribe ();
 122
 123   // = Event/state/action handlers.
 124   /// Receive a event from stdin and send it to the gateway.
 125   int transmit_stdin ();
 126
 127   /// Action that receives the route id.
 128   int await_connection_id ();
 129
 130   /// Action that receives events.
 131   int await_events ();
 132
 133   /// Pointer-to-member-function for the current action to run in this
 134   /// state.  This points to one of the preceding 3 methods.
 135   int (Peer_Handler::*do_action_)();
 136
 137   /// Connection ID of the peer, which is obtained from the gatewayd.
 138   CONNECTION_ID connection_id_;
 139
 140   /// Keep track of event fragments that arrive in non-blocking recv's
 141   /// from the gatewayd.
 142   ACE_Message_Block *msg_frag_;
 143
 144   /// The total number of bytes sent/received to the gatewayd thus far.
 145   size_t total_bytes_;
 146
 147   /// Used to call register_stdin_handle only once.  Otherwise, thread
 148   /// leak will occur on Win32.
 149   int first_time_;
 150 };
 151
 152 #if defined ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION_EXPORT
 153 template class ACE_Svc_Export ACE_Acceptor<Peer_Handler, ACE_SOCK_ACCEPTOR>;
 154 #endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION_EXPORT */
 155
 156 /**
 157  * @class Peer_Acceptor
 158  *
 159  * @brief Passively accept connections from gatewayd and dynamically
 160  * create a new <Peer_Handler> object to communicate with the
 161  * gatewayd.
 162  */
 163 class ACE_Svc_Export Peer_Acceptor : public ACE_Acceptor<Peer_Handler, ACE_SOCK_ACCEPTOR>
 164 {
 165 public:
 166   /// Default initialization.
 167   Peer_Acceptor ();
 168
 169   ///  the <Peer_Acceptor>.
 170   int start (u_short);
 171
 172   /// Terminate the <Peer_Acceptor>.
 173   int close ();
 174
 175   /// Factory method that creates a <Peer_Handler> just once.
 176   virtual int make_svc_handler (Peer_Handler *&);
 177
 178 private:
 179   /// Factor out common code for initializing the <Peer_Acceptor>.
 180   int open_acceptor (u_short port);
 181
 182   /// Pointer to <Peer_Handler> allocated just once.
 183   Peer_Handler *peer_handler_;
 184
 185   /// Our acceptor addr.
 186   ACE_INET_Addr addr_;
 187
 188   typedef ACE_Acceptor<Peer_Handler, ACE_SOCK_ACCEPTOR> inherited;
 189 };
 190
 191 #if defined ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION_EXPORT
 192 template class ACE_Svc_Export ACE_Connector<Peer_Handler, ACE_SOCK_CONNECTOR>;
 193 #endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION_EXPORT */
 194
 195 /**
 196  * @class Peer_Connector
 197  *
 198  * @brief Actively establish connections with gatewayd and dynamically
 199  * create a new <Peer_Handler> object to communicate with the
 200  * gatewayd.
 201  */
 202 class ACE_Svc_Export Peer_Connector : public ACE_Connector<Peer_Handler, ACE_SOCK_CONNECTOR>
 203 {
 204 public:
 205   /**
 206    * Initialize the <Peer_Connector>.  NOTE:  the arguments are
 207    * ignored.  They are only provided to avoid a compiler warning
 208    * about hiding the virtual function ACE_Connector<Peer_Handler,
 209    * ACE_SOCK_CONNECTOR>::open(ACE_Reactor*, int).
 210    */
 211   int open (ACE_Reactor * = 0, int = 0);
 212
 213 private:
 214   /// Factor out common code for initializing the <Peer_Connector>.
 215   int open_connector (Peer_Handler *&ph, u_short port);
 216
 217   /// Consumer <Peer_Handler> that is connected to a gatewayd.
 218   Peer_Handler *consumer_peer_handler_;
 219
 220   /// Supplier <Peer_Handler> that is connected to a gatewayd.
 221   Peer_Handler *supplier_peer_handler_;
 222 };
 223
 224 /**
 225  * @class Peer_Factory
 226  *
 227  * @brief A factory class that actively and/or passively establishes
 228  * connections with the gatewayd.
 229  */
 230 class ACE_Svc_Export Peer_Factory : public ACE_Service_Object
 231 {
 232 public:
 233   // = Dynamic initialization and termination hooks from <ACE_Service_Object>.
 234
 235   /// Initialize the acceptor and connector.
 236   virtual int init (int argc, ACE_TCHAR *argv[]);
 237
 238   /// Perform termination activities.
 239   virtual int fini ();
 240
 241   /// Return info about this service.
 242   virtual int info (ACE_TCHAR **, size_t) const;
 243
 244   /// Handle various signals (e.g., SIGPIPE, SIGINT, and SIGQUIT).
 245   virtual int handle_signal (int signum, siginfo_t *, ucontext_t *);
 246
 247 private:
 248   /// Pointer to an instance of our <Peer_Acceptor> that's used to
 249   /// accept connections and create Consumers.
 250   Peer_Acceptor consumer_acceptor_;
 251
 252   /// Pointer to an instance of our <Peer_Acceptor> that's used to
 253   /// accept connections and create Suppliers.
 254   Peer_Acceptor supplier_acceptor_;
 255
 256   /**
 257    * An instance of our <Peer_Connector>.  Note that one
 258    * <Peer_Connector> is used to establish <Peer_Handler>s for both
 259    * Consumers and Suppliers.
 260    */
 261   Peer_Connector connector_;
 262 };
 263
 264 #endif /* PEER_H */