talker.cpp

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

// Talker.cpp,v 4.8 2003/11/09 04:12:08 dhinton Exp

// ============================================================================
//
// = LIBRARY
//    examples
//
// = FILENAME
//    Talker.cpp
//
// = DESCRIPTION
//
//    This test application tests a wide range of events that can be
//    demultiplexed using various ACE utilities.  Events used include
//    ^C events, reading from STDIN, vanilla Win32 events, thread
//    exits, Reactor notifications, proactive reads, and proactive
//    writes.
//
//    The proactive I/O events are demultiplexed by the ACE_Proactor.
//    The thread exits, notications, and vanilla Win32 events are
//    demultiplexed by the ACE_Reactor.  To enable a single thread
//    to run all these events, the Proactor is integrated with the
//    Reactor.
//
//    The test application prototypes a simple talk program.  Two
//    instances of the application connect.  Input from either console
//    is displayed on the others console also.  Because of the evils
//    of Win32 STDIN, a separate thread is used to read from STDIN.
//    To test the Proactor and Reactor, I/O between the remote
//    processes is performed proactively and interactions between the
//    STDIN thread and the main thread are performed reactively.
//
//    The following description of the test application is in two
//    parts.  The participants section explains the main components
//    involved in the application.  The collaboration section
//    describes how the partipants interact in response to the
//    multiple event types which occur.
//
//    The Reactor test application has the following participants:
//
//    . Reactor -- The Reactor demultiplexes Win32 "waitable"
//    events using WaitForMultipleObjects.
//
//    . Proactor -- The proactor initiates and demultiplexes
//    overlapped I/O operations.  The Proactor registers with the
//    Reactor so that a single-thread can demultiplex all
//    application events.
//
//    . STDIN_Handler -- STDIN_Handler is an Active Object which reads
//    from STDIN and forwards the input to the Peer_Handler.  This
//    runs in a separate thread to make the test more interesting.
//    However, STDIN is "waitable", so in general it can be waited on
//    by the ACE Reactor, thanks MicroSlush!
//
//    . Peer_Handler -- The Peer_Handler connects to another instance
//    of test_reactor.  It Proactively reads and writes data to the
//    peer.  When the STDIN_Handler gives it messages, it fowards them
//    to the remote peer.  When it receives messages from the remote
//    peer, it prints the output to the console.
//
//    The collaborations of the participants are as follows:
//
//    . Initialization
//
//      Peer_Handler -- connects to the remote peer.  It then begins
//      proactively reading from the remote connection.  Note that it
//      will be notified by the Proactor when a read completes.  It
//      also registers a notification strategy with message queue so
//      that it is notified when the STDIN_Handler posts a message
//      onto the queue.
//
//      STDIN_Handler -- STDIN_Handler registers a signal handler for
//      SIGINT.  This just captures the exception so that the kernel
//      doesn't kill our process; We want to exit gracefully.  It also
//      creates an Exit_Hook object which registers the
//      STDIN_Handler's thread handle with the Reactor.  The
//      Exit_Hook will get called back when the STDIN_Handler thread
//      exits.  After registering these, it blocks reading from STDIN.
//
//      Proactor -- is registered with the Reactor.
//
//      The main thread of control waits in the Reactor.
//
//    . STDIN events -- When the STDIN_Handler thread reads from
//    STDIN, it puts the message on Peer_Handler's message queue.  It
//    then returns to reading from STDIN.
//
//    . Message enqueue -- The Reactor thread wakes up and calls
//    Peer_Handler::handle_output.  The Peer_Handler then tries to
//    dequeue a message from its message queue.  If it can, the
//    message is Proactively sent to the remote peer.  Note that the
//    Peer_Handler will be notified with this operation is complete.
//    The Peer_Handler then falls back into the Reactor event loop.
//
//    . Send complete event -- When a proactive send is complete, the
//    Proactor is notified by the Reactor.  The Proactor, in turn,
//    notifies the Peer_Handler.  The Peer_Handler then checks for
//    more messages from the message queue.  If there are any, it
//    tries to send them.  If there are not, it returns to the
//    Reactor event loop.
//
//    . Read complete event -- When a proactive read is complete (the
//    Peer_Handler initiated a proactive read when it connected to the
//    remote peer), the Proactor is notified by the Reactor.  The
//    Proactor, in turn notifies the Peer_Handler.  If the read was
//    successful the Peer_Handler just displays the received msg to
//    the console and reinvokes a proactive read from the network
//    connection.  If the read failed (i.e. the remote peer exited),
//    the Peer_Handler sets a flag to end the event loop and returns.
//    This will cause the application to exit.
//
//    . ^C events -- When the user types ^C at the console, the
//    STDIN_Handler's signal handler will be called.  It does nothing,
//    but as a result of the signal, the STDIN_Handler thread will
//    exit.
//
//    . STDIN_Handler thread exits -- The Exit_Hook will get called
//    back from the Reactor.  Exit_Hook::handle_signal sets a flag
//    to end the event loop and returns.  This will cause the
//    application to exit.
//
//
//    To run example, start an instance of the test with an optional
//    local port argument (as the acceptor). Start the other instance
//    with -h <hostname> and -p <server port>. Type in either the
//    client or server windows and your message should show up in the
//    other window.  Control C to exit.
//
// = AUTHOR
//    Tim Harrison
//    Irfan Pyarali
//
// ============================================================================

#include "ace/OS_main.h"

#if defined (ACE_WIN32)

#include "ace/Reactor.h"
#include "ace/Reactor_Notification_Strategy.h"
#include "ace/WIN32_Proactor.h"
#include "ace/Proactor.h"
#include "ace/SOCK_Connector.h"
#include "ace/SOCK_Acceptor.h"
#include "ace/Get_Opt.h"
#include "ace/Service_Config.h"
#include "ace/Task.h"

ACE_RCSID(WFMO_Reactor, Talker, "Talker.cpp,v 4.8 2003/11/09 04:12:08 dhinton Exp")

typedef ACE_Task<ACE_MT_SYNCH> MT_TASK;

class Peer_Handler : public MT_TASK, public ACE_Handler
// = TITLE
//     Connect to a server.  Receive messages from STDIN_Handler
//     and forward them to the server using proactive I/O.
{
public:
  // = Initialization methods.
  Peer_Handler (int argc, ACE_TCHAR *argv[]);
  ~Peer_Handler (void);

  int open (void * =0);
  // This method creates the network connection to the remote peer.
  // It does blocking connects and accepts depending on whether a
  // hostname was specified from the command line.

  virtual void handle_read_stream (const ACE_Asynch_Read_Stream::Result &result);
  // This method will be called when an asynchronous read completes on a stream.
  // The remote peer has sent us something.  If it succeeded, print
  // out the message and reinitiate a read.  Otherwise, fail.  In both
  // cases, delete the message sent.

  virtual void handle_write_stream (const ACE_Asynch_Write_Stream::Result &result);
  // This method will be called when an asynchronous write completes on a strea_m.
  // One of our asynchronous writes to the remote peer has completed.
  // Make sure it succeeded and then delete the message.

  virtual ACE_HANDLE handle (void) const;
  // Get the I/O handle used by this <handler>. This method will be
  // called by the ACE_Asynch_* classes when an ACE_INVALID_HANDLE is
  // passed to <open>.

  void handle (ACE_HANDLE);
  // Set the ACE_HANDLE value for this Handler.

  virtual int handle_close (ACE_HANDLE, ACE_Reactor_Mask);
  // We've been removed from the Reactor.

  virtual int handle_output (ACE_HANDLE fd);
  // Called when output events should start.  Note that this is
  // automatically invoked by the
  // <ACE_Reactor_Notificiation_Strategy>.

private:
  ACE_SOCK_Stream stream_;
  // Socket that we have connected to the server.

  ACE_Reactor_Notification_Strategy strategy_;
  // The strategy object that the reactor uses to notify us when
  // something is added to the queue.

  // = Remote peer info.
  ACE_TCHAR *host_;
  // Name of remote host.

  u_short port_;
  // Port number for remote host.

  ACE_Asynch_Read_Stream rd_stream_;
  // Read stream

  ACE_Asynch_Write_Stream wr_stream_;
  // Write stream

  ACE_Message_Block mb_;
  // Message Block for reading from the network
};

class STDIN_Handler : public ACE_Task<ACE_NULL_SYNCH>
// = TITLE
//    Active Object.  Reads from STDIN and passes message blocks to
//    the peer handler.
{
public:
  STDIN_Handler (MT_TASK &ph);
  // Initialization.

  virtual int open (void * = 0);
  // Activate object.

  virtual int close (u_long = 0);
  // Shut down.

  int svc (void);
  // Thread runs here as an active object.

  int handle_close (ACE_HANDLE,
                    ACE_Reactor_Mask);

private:
  static void handler (int signum);
  // Handle a ^C.  (Do nothing, this just illustrates how we can catch
  // signals along with the other things).

  void register_thread_exit_hook (void);
  // Helper function to register with the Reactor for thread exit.

  virtual int handle_signal (int index, siginfo_t *, ucontext_t *);
  // The STDIN thread has exited.  This means the user hit ^C.  We can
  // end the event loop.

  MT_TASK &ph_;
  // Send all input to ph_.

  ACE_HANDLE thr_handle_;
  // Handle of our thread.
};

Peer_Handler::Peer_Handler (int argc, ACE_TCHAR *argv[])
  : host_ (0),
    port_ (ACE_DEFAULT_SERVER_PORT),
    strategy_ (ACE_Reactor::instance (),
               this,
               ACE_Event_Handler::WRITE_MASK),
    mb_ (BUFSIZ)
{
  // This code sets up the message to notify us when a new message is
  // added to the queue.  Actually, the queue notifies Reactor which
  // then notifies us.
  this->msg_queue ()->notification_strategy (&this->strategy_);

  ACE_Get_Opt get_opt (argc, argv, ACE_TEXT("h:p:"));
  int c;

  while ((c = get_opt ()) != EOF)
    {
      switch (c)
        {
        case 'h':
          host_ = get_opt.opt_arg ();
          break;
        case 'p':
          port_ = ACE_OS::atoi (get_opt.opt_arg ());
          break;
        }
    }
}

Peer_Handler::~Peer_Handler (void)
{
}

// This method creates the network connection to the remote peer.  It
// does blocking connects and accepts depending on whether a hostname
// was specified from the command line.