tcp.cpp

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

// TCP.cpp,v 5.18 2003/11/04 05:21:31 dhinton Exp

#include "TCP.h"
#include "AVStreams_i.h"

#include "tao/debug.h"
#include "ace/OS_NS_strings.h"

//------------------------------------------------------------
// TAO_AV_TCP_Transport
//------------------------------------------------------------

TAO_AV_TCP_Transport::TAO_AV_TCP_Transport (void)
  :handler_ (0)
{
}

TAO_AV_TCP_Transport::TAO_AV_TCP_Transport (TAO_AV_TCP_Flow_Handler *handler)
  :handler_ (handler)
{
}

TAO_AV_TCP_Transport::~TAO_AV_TCP_Transport (void)
{
}

int
TAO_AV_TCP_Transport::open (ACE_Addr * /*address*/)
{
  return 0;
}

int
TAO_AV_TCP_Transport::close (void)
{
  return 0;
}

int
TAO_AV_TCP_Transport::mtu (void)
{
  return -1;
}

ACE_Addr*
TAO_AV_TCP_Transport::get_peer_addr (void)
{
  return 0;
}

ssize_t
TAO_AV_TCP_Transport::send (const ACE_Message_Block *mblk, ACE_Time_Value *)
{
  // For the most part this was copied from GIOP::send_request and
  // friends.

  iovec iov[ACE_IOV_MAX];
  int iovcnt = 0;
  ssize_t n = 0;
  ssize_t nbytes = 0;

  for (const ACE_Message_Block *i = mblk;
       i != 0;
       i = i->cont ())
    {
      // Make sure there is something to send!
      if (i->length () > 0)
        {
          iov[iovcnt].iov_base = i->rd_ptr ();
          iov[iovcnt].iov_len  = ACE_static_cast (u_long, i->length ());
          iovcnt++;

          // The buffer is full make a OS call.  @@ TODO this should
          // be optimized on a per-platform basis, for instance, some
          // platforms do not implement writev() there we should copy
          // the data into a buffer and call send_n(). In other cases
          // there may be some limits on the size of the iovec, there
          // we should set ACE_IOV_MAX to that limit.
          if (iovcnt == ACE_IOV_MAX)
            {
              n = this->handler_->peer ().sendv_n ((const iovec *) iov,
                                                   iovcnt);
              if (n < 1)
                return n;

              nbytes += n;
              iovcnt = 0;
            }
        }
    }

  // Check for remaining buffers to be sent!
  if (iovcnt != 0)
    {
      n = this->handler_->peer ().sendv_n ((const iovec *) iov,
                                           iovcnt);
      if (n < 1)
        return n;

      nbytes += n;
    }

  return nbytes;
}

ssize_t
TAO_AV_TCP_Transport::send (const char *buf,
                            size_t len,
                            ACE_Time_Value *)
{
  return this->handler_->peer ().send_n (buf, len);
}

ssize_t
TAO_AV_TCP_Transport::send (const iovec *iov,
                          int iovcnt,
                          ACE_Time_Value *)
{
  return this->handler_->peer ().sendv_n ((const iovec *) iov,
                                          iovcnt);
}

ssize_t
TAO_AV_TCP_Transport::recv (char *buf,
                          size_t len,
                          ACE_Time_Value *)
{
  return this->handler_->peer ().recv (buf, len);
}

ssize_t
TAO_AV_TCP_Transport::recv (char *buf,
                            size_t len,
                            int flags,
                            ACE_Time_Value *)
{
  return this->handler_->peer ().recv (buf,
                                       len,
                                       flags);
}

ssize_t
TAO_AV_TCP_Transport::recv (iovec *iov,
                            int iovcnt,
                            ACE_Time_Value *)
{
  return handler_->peer ().recvv_n (iov, iovcnt);
}

//------------------------------------------------------------
// TAO_AV_TCP_Protocol_Factory
//------------------------------------------------------------

TAO_AV_TCP_Factory::TAO_AV_TCP_Factory (void)
{
}

TAO_AV_TCP_Factory::~TAO_AV_TCP_Factory (void)
{
}

int
TAO_AV_TCP_Factory::init (int /* argc */,
                          char * /* argv */ [])
{
  return 0;
}

int
TAO_AV_TCP_Factory::match_protocol (const char *protocol_string)
{
  if (ACE_OS::strcasecmp (protocol_string,"TCP") == 0)
    return 1;
  return 0;
}

TAO_AV_Acceptor*
TAO_AV_TCP_Factory::make_acceptor (void)
{
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"TAO_AV_TCP_Factory::make_acceptor "));
  TAO_AV_Acceptor *acceptor = 0;
  ACE_NEW_RETURN (acceptor,
                  TAO_AV_TCP_Acceptor,
                  0);
  return acceptor;
}

TAO_AV_Connector*
TAO_AV_TCP_Factory::make_connector (void)
{
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"TAO_AV_TCP_Factory::make_connector "));
  TAO_AV_Connector *connector = 0;
  ACE_NEW_RETURN (connector,
                  TAO_AV_TCP_Connector,
                  0);
  return connector;
}

//------------------------------------------------------------
// TAO_AV_TCP_Object
//------------------------------------------------------------

int
TAO_AV_TCP_Object::handle_input (void)
{
  int n = this->transport_->recv (this->frame_.rd_ptr (),
                                  this->frame_.size ());
  if (n == -1)
    ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_TCP_Flow_Handler::handle_input recv failed\n"),-1);
  if (n == 0)
    ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_TCP_Flow_Handler::handle_input connection closed\n"),-1);
  this->frame_.wr_ptr (this->frame_.rd_ptr () + n);

  return this->callback_->receive_frame (&this->frame_);
}

int
TAO_AV_TCP_Object::send_frame (ACE_Message_Block *frame,
                               TAO_AV_frame_info * /*frame_info*/)
{
  int result = this->transport_->send (frame);
  return result;
}

int
TAO_AV_TCP_Object::send_frame (const iovec *iov,
                               int iovcnt,
                               TAO_AV_frame_info * /*frame_info*/)
{
  return this->transport_->send (iov,iovcnt);
}

int
TAO_AV_TCP_Object::send_frame (const char*buf,
                               size_t len)
{
  int result = this->transport_->send (buf, len, 0);
  return result;
}


TAO_AV_TCP_Object::TAO_AV_TCP_Object (TAO_AV_Callback *callback,
                                      TAO_AV_Transport *transport)
  :TAO_AV_Protocol_Object (callback,transport)
{
  // @@ Is this a good size?
  this->frame_.size (BUFSIZ);
}

TAO_AV_TCP_Object::~TAO_AV_TCP_Object (void)
{
  // No-op
}
int
TAO_AV_TCP_Object::destroy (void)
{
  this->callback_->handle_destroy ();
  delete this;
  return 0;
}

//------------------------------------------------------------
// TAO_AV_TCP_Flow_Factory
//------------------------------------------------------------
TAO_AV_TCP_Flow_Factory::TAO_AV_TCP_Flow_Factory (void)
{
}

TAO_AV_TCP_Flow_Factory::~TAO_AV_TCP_Flow_Factory (void)
{
}

int
TAO_AV_TCP_Flow_Factory::init (int /* argc */,
                               char * /* argv */ [])
{
  return 0;
}

int
TAO_AV_TCP_Flow_Factory::match_protocol (const char *flow_string)
{
  if (ACE_OS::strcasecmp (flow_string,"TCP") == 0)
    return 1;
  return 0;
}

TAO_AV_Protocol_Object*
TAO_AV_TCP_Flow_Factory::make_protocol_object (TAO_FlowSpec_Entry *entry,
                                               TAO_Base_StreamEndPoint *endpoint,
                                               TAO_AV_Flow_Handler *handler,
                                               TAO_AV_Transport *transport)
{
  TAO_AV_Callback *callback = 0;
  if ( endpoint->get_callback (entry->flowname (), callback) ) {
    ACE_ERROR_RETURN ((LM_ERROR, "(%N,%l) Invalid callback\n"), 0);
  }

  TAO_AV_TCP_Object *object = 0;
  ACE_NEW_RETURN (object,
                  TAO_AV_TCP_Object (callback,
                                     transport),
                  0);
  callback->open (object,
                  handler);
  endpoint->set_protocol_object (entry->flowname (),
                                 object);
  return object;
}

//------------------------------------------------------------
// TAO_AV_TCP_Base_Connector
//------------------------------------------------------------

int
TAO_AV_TCP_Base_Connector::connector_open (TAO_AV_TCP_Connector *connector,
                                           ACE_Reactor *reactor)
{
  this->connector_ = connector;
  this->reactor_ = reactor;

  int result = ACE_Connector <TAO_AV_TCP_Flow_Handler,ACE_SOCK_CONNECTOR>::open (reactor);
  if (result < 0)
    ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_TCP_Base_Connector::open failed\n"),-1);
  return 0;
}

int
TAO_AV_TCP_Base_Connector::make_svc_handler (TAO_AV_TCP_Flow_Handler *&tcp_handler)
{
  int result =
    this->connector_->make_svc_handler (tcp_handler);
  if (result < 0)
    return result;
  tcp_handler->reactor (this->reactor_);
  return 0;
}

int
TAO_AV_TCP_Base_Connector::connector_connect (TAO_AV_TCP_Flow_Handler *&handler,
                                              const ACE_INET_Addr &remote_addr)
{
  int result = ACE_Connector <TAO_AV_TCP_Flow_Handler,ACE_SOCK_CONNECTOR>::connect (handler,
                                                                                    remote_addr);
  if (result < 0)
    ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_TCP_Base_Connector::connect failed\n"),-1);
  return 0;
}

//------------------------------------------------------------
// TAO_AV_TCP_Connector
//------------------------------------------------------------
TAO_AV_TCP_Connector::TAO_AV_TCP_Connector (void)
{
}

TAO_AV_TCP_Connector::~TAO_AV_TCP_Connector (void)
{
}

int
TAO_AV_TCP_Connector::make_svc_handler (TAO_AV_TCP_Flow_Handler *&tcp_handler)