to myself 的分类学习日志

/* TCP_Peer_Event_Handler.h
 */ 
#ifndef _TCP_PEER_EVENT_HANDLER_H_
#define _TCP_PEER_EVENT_HANDLER_H_

#include "ace/Message_Block.h"
#include "ace/SOCK_Stream.h"
#include "ace/Svc_Handler.h"


class TCP_Peer_Event_Handler :
  public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>
{
  typedef ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH> super;

public:
    TCP_Peer_Event_Handler();
    virtual ~TCP_Peer_Event_Handler();

public:
   /// Activate the client handler.  This is typically called by the
  /// ACE_Acceptor or ACE_Connector.
  virtual int open (void * = 0);

  // Called when input is available from the client.
  virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE);

  // Called when output is possible.
  virtual int handle_output (ACE_HANDLE fd = ACE_INVALID_HANDLE);

  // Called when this handler is removed from the ACE_Reactor.
  virtual int handle_close (ACE_HANDLE handle,
                            ACE_Reactor_Mask close_mask);
};


#endif // _TCP_PEER_EVENT_HANDLER_H_

/* TCP_Peer_Event_Handler.cpp
 */ 
#include "TCP_Peer_Event_Handler.h"

#define MAXHOSTNAMELEN   255


TCP_Peer_Event_Handler::TCP_Peer_Event_Handler()
{
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) TCP_Peer_Event_Handler constructor!\n")));
    
    this->reactor(get_reactor());

    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%P|%t) TCP_Peer_Event_Handler() Using reactor 0x%x\n"),
                this->reactor()));
}


TCP_Peer_Event_Handler::~TCP_Peer_Event_Handler()
{
    ACE_DEBUG ((LM_DEBUG, ACE_TEXT("(%P|%t) TCP_Peer_Event_Handler destructor!\n")));
}


int TCP_Peer_Event_Handler::open(void *p)
{
    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%P|%t) open() Using reactor 0x%x\n"),
                this->reactor()));

  if (super::open (p) == -1)
    return -1;

  ACE_TCHAR peer_name[MAXHOSTNAMELEN];
  ACE_INET_Addr peer_addr;
  if (this->peer ().get_remote_addr (peer_addr) == 0 &&
      peer_addr.addr_to_string (peer_name, MAXHOSTNAMELEN) == 0)
    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%P|%t) Connection from %s\n"),
                peer_name));
  return 0;
}


int TCP_Peer_Event_Handler::handle_input(ACE_HANDLE)
{
    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%P|%t) handle_input() Using reactor 0x%x\n"),
                this->reactor()));


  const size_t INPUT_SIZE = 4096;
  char buffer[INPUT_SIZE];
  ssize_t recv_cnt, send_cnt;

  recv_cnt = this->peer ().recv (buffer, sizeof(buffer));
  if (recv_cnt <= 0)
    {
      ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("(%P|%t) Connection closed\n")));
      return -1;
    }
  else
  {
    buffer[recv_cnt] = '\0';
    ACE_DEBUG ((LM_DEBUG,
                  ACE_TEXT ("(%P|%t) Recv %s\n"), buffer));
  }

  send_cnt =
    this->peer ().send (buffer,
                        ACE_static_cast (size_t, recv_cnt));
  if (send_cnt == recv_cnt)
    return 0;
  if (send_cnt == -1 && ACE_OS::last_error () != EWOULDBLOCK)
    ACE_ERROR_RETURN ((LM_ERROR,
                       ACE_TEXT ("(%P|%t) %p\n"),
                       ACE_TEXT ("send")),
                      0);
  if (send_cnt == -1)
    send_cnt = 0;
  ACE_Message_Block *mb;
  size_t remaining =
    ACE_static_cast (size_t, (recv_cnt - send_cnt));
  ACE_NEW_RETURN
    (mb, ACE_Message_Block (&buffer[send_cnt], remaining), -1);
  int output_off = this->msg_queue ()->is_empty ();
  ACE_Time_Value nowait (ACE_OS::gettimeofday ());
  if (this->putq (mb, &nowait) == -1)
    {
      ACE_ERROR ((LM_ERROR,
                  ACE_TEXT ("(%P|%t) %p; discarding data\n"),
                  ACE_TEXT ("enqueue failed")));
      mb->release ();
      return 0;
    }
  if (output_off)
    return this->reactor ()->register_handler
      (this, ACE_Event_Handler::WRITE_MASK);
  return 0;
}


int TCP_Peer_Event_Handler::handle_output(ACE_HANDLE)
{
    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%P|%t) handle_output() Using reactor 0x%x\n"),
                this->reactor()));


  ACE_Message_Block *mb;
  ACE_Time_Value nowait (ACE_OS::gettimeofday ());
  while (-1 != this->getq (mb, &nowait))
    {
      ssize_t send_cnt =
        this->peer ().send (mb->rd_ptr (), mb->length ());
      if (send_cnt == -1)
        ACE_ERROR ((LM_ERROR,
                    ACE_TEXT ("(%P|%t) %p\n"),
                    ACE_TEXT ("send")));
      else
        mb->rd_ptr (ACE_static_cast (size_t, send_cnt));
      if (mb->length () > 0)
        {
          this->ungetq (mb);
          break;
        }
      mb->release ();
    }
  return (this->msg_queue ()->is_empty ()) ? -1 : 0;
}


int TCP_Peer_Event_Handler::handle_close (ACE_HANDLE h, ACE_Reactor_Mask mask)
{
    ACE_DEBUG ((LM_DEBUG,
                ACE_TEXT ("(%P|%t) handle_close() Using reactor 0x%x\n"),
                this->reactor()));


  if (mask == ACE_Event_Handler::WRITE_MASK)
    return 0;
  return super::handle_close (h, mask);
}

/* ACE_TCP_Server.cpp
 */ 
#include "TCP_Peer_Event_Handler.h"
#include "ace/Log_Msg.h"
#include "ace/INET_Addr.h"
#include "ace/SOCK_Acceptor.h"
#include "ace/Reactor.h"
#include "ace/Acceptor.h"
#include <ace/TP_Reactor.h>
#include <ace/Dev_Poll_Reactor.h>


typedef ACE_Acceptor<TCP_Peer_Event_Handler, ACE_SOCK_ACCEPTOR> 
  TCP_Client_Acceptor;


int ACE_TMAIN (int, ACE_TCHAR *[])
{
  ACE_INET_Addr port_to_listen("8112");
  TCP_Client_Acceptor acceptor;
  if (acceptor.open (port_to_listen) == -1)
    return 1;

  ACE_Reactor::instance ()->run_reactor_event_loop ();

  return (0);
}

/* ACE_TCP_Client.cpp
 */ 
#include "TCP_Peer_Event_Handler.h"

typedef ACE_Connector<TCP_Peer_Event_Handler, ACE_SOCK_CONNECTOR>
  TCP_Client_Connector;

int ACE_TMAIN (int, ACE_TCHAR *[])
{
  ACE_INET_Addr port_to_connect ("8112", ACE_LOCALHOST);
  TCP_Client_Connector connector;
  TCP_Peer_Event_Handler client;
  if (connector.connect (&client, port_to_connect) == -1)
    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"),
                       ACE_TEXT ("connect")), 1);

  ACE_Reactor::instance ()->run_reactor_event_loop ();
  return (0);
}