udp.cpp

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

// UDP.cpp,v 5.30 2003/11/04 05:21:31 dhinton Exp

#include "UDP.h"
#include "AVStreams_i.h"
#include "MCast.h"

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

#if !defined (__ACE_INLINE__)
#include "UDP.i"
#endif /* __ACE_INLINE__ */

ACE_RCSID (AV,
           UDP,
           "UDP.cpp,v 5.30 2003/11/04 05:21:31 dhinton Exp")

//------------------------------------------------------------
// TAO_AV_UDP_Flow_Handler
//------------------------------------------------------------

TAO_AV_UDP_Flow_Handler::TAO_AV_UDP_Flow_Handler (void)
{
  ACE_NEW (this->transport_,
           TAO_AV_UDP_Transport (this));
}

TAO_AV_UDP_Flow_Handler::~TAO_AV_UDP_Flow_Handler (void)
{
  // remove the event handler from the reactor.
  TAO_AV_CORE::instance()->reactor ()->remove_handler (this->event_handler(),
                                              ACE_Event_Handler::READ_MASK);

  // close the socket.
  this->close ();
  delete this->transport_;
}

TAO_AV_Transport *
TAO_AV_UDP_Flow_Handler::transport (void)
{
  return this->transport_;
}

int
TAO_AV_UDP_Flow_Handler::handle_input (ACE_HANDLE /*fd*/)
{
  return this->protocol_object_->handle_input ();
}

int
TAO_AV_UDP_Flow_Handler::handle_timeout (const ACE_Time_Value &tv,
                                         const void *arg)
{
  return TAO_AV_Flow_Handler::handle_timeout (tv,arg);
}

int
TAO_AV_UDP_Flow_Handler::set_remote_address (ACE_Addr *address)
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,"TAO_AV_UDP_Flow_Handler::set_remote_address\n"));

  ACE_INET_Addr *inet_addr = ACE_dynamic_cast (ACE_INET_Addr*,
                                               address);
  this->peer_addr_ = *inet_addr;
  TAO_AV_UDP_Transport *transport = ACE_dynamic_cast (TAO_AV_UDP_Transport*,
                                                      this->transport_);

  return transport->set_remote_address (*inet_addr);
}


ACE_HANDLE
TAO_AV_UDP_Flow_Handler::get_handle (void) const
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "TAO_AV_UDP_Flow_Handler::get_handle:%d\n",
                this->sock_dgram_.get_handle ()));

  return this->sock_dgram_.get_handle () ;
}

int
TAO_AV_UDP_Flow_Handler::change_qos(AVStreams::QoS qos)
{
  if( TAO_debug_level > 0 )
    {
      ACE_DEBUG ((LM_DEBUG,
                  "(%N,%l) TAO_AV_UDP_Flow_Handler::change_qos\n"));
    }

  unsigned int i=0;

  int ret = 0;
  CORBA::Long dscp = 0;
  CORBA::Long ecn = 0;
  int dscp_flag=0;
  for(i=0; i < qos.QoSParams.length(); i++)
    {

      if( ACE_OS::strcmp( qos.QoSParams[i].property_name.in(), "Diffserv_Codepoint") == 0)
        {
          qos.QoSParams[i].property_value >>= dscp;
          dscp_flag=1;
          // DSCP value can only be 6 bits wide
          if(!((dscp >= 0) && (dscp <= 63)))
            {
              dscp_flag = 0;
              ACE_DEBUG((LM_DEBUG, "(%N,%l) ECN value can only be (0-3) not %d\n", ecn));
              return -1;
            }
        }

      if( ACE_OS::strcmp( qos.QoSParams[i].property_name.in(), "ECN") == 0)
        {
          qos.QoSParams[i].property_value >>= ecn;
          // ECN value can only occupy bits 6 and 7 of the
          // IP Diffserv byte
          if(!((ecn >= 0) && (ecn <= 3)))
            {
              ACE_DEBUG((LM_DEBUG, "(%N,%l) ECN value can only be (0-3) not %d\n", ecn));
              ecn = 0;
            }

        }
    }
      // Set the Diffserv byte only if we specified
      // the Diffserv Codepoint (DSCP) or ECN via QoSParams
      // passed into this method
      if(dscp_flag || ecn)
        {
          int tos;
          tos = (int)(dscp << 2);
          if(ecn)
            {
              tos |= ecn;
            }
          ret = sock_dgram_.set_option(IPPROTO_IP, IP_TOS, (int *)&tos , (int)sizeof(tos));

          if(TAO_debug_level > 1)
            {
              ACE_DEBUG((LM_DEBUG, "(%N,%l) set tos: ret: %d\n", ret));
            }
        }

      if(TAO_debug_level > 1)
        {
           if(ret < 0 )
             {
                ACE_DEBUG((LM_DEBUG, "(%N,%l) errno: %p\n"));
             }
        }
      return ret;
}

//------------------------------------------------------------
// TAO_AV_UDP_Transport
//------------------------------------------------------------

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

TAO_AV_UDP_Transport::TAO_AV_UDP_Transport (TAO_AV_UDP_Flow_Handler *handler)
  :handler_ (handler),
   addr_ (0)
{
}

TAO_AV_UDP_Transport::~TAO_AV_UDP_Transport (void)
{
}

int
TAO_AV_UDP_Transport::set_remote_address (const ACE_INET_Addr &address)
{
  this->peer_addr_ = address;
  return 0;
}

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

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

int
TAO_AV_UDP_Transport::mtu (void)
{
  return 65535;
}

ACE_Addr*
TAO_AV_UDP_Transport::get_peer_addr (void)
{
  return &this->peer_addr_;
}

ssize_t
TAO_AV_UDP_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_->get_socket ()->send ((const iovec *) iov,
                                                        iovcnt,
                                                        this->peer_addr_);

              if (n < 1)
                return n;

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

  // Check for remaining buffers to be sent!
  if (iovcnt != 0)
    {
      n = this->handler_->get_socket ()->send ((const iovec *) iov,
                                               iovcnt,
                                               this->peer_addr_);

      if (n < 1)
        return n;

      nbytes += n;
    }

  return nbytes;
}

ssize_t
TAO_AV_UDP_Transport::send (const char *buf,
                            size_t len,
                            ACE_Time_Value *)
{
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"TAO_AV_UDP_Transport::send "));
  char addr [BUFSIZ];
  this->peer_addr_.addr_to_string (addr,BUFSIZ);
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"to %s\n",addr));

  return this->handler_->get_socket ()->send (buf, len,this->peer_addr_);
}

ssize_t
TAO_AV_UDP_Transport::send (const iovec *iov,
                          int iovcnt,
                          ACE_Time_Value *)
{
  return this->handler_->get_socket ()->send ((const iovec *) iov,
                                              iovcnt,
                                              this->peer_addr_);

}

ssize_t
TAO_AV_UDP_Transport::recv (char *buf,
                            size_t len,
                            ACE_Time_Value *)
{
  return this->handler_->get_socket ()->recv (buf, len,this->peer_addr_);
}

ssize_t
TAO_AV_UDP_Transport::recv (char *buf,
                            size_t len,
                            int flags,
                            ACE_Time_Value *timeout)
{
  return this->handler_->get_socket ()->recv (buf,
                                              len,
                                              this->peer_addr_,
                                              flags,
                                              timeout);
}

ssize_t
TAO_AV_UDP_Transport::recv (iovec *iov,
                            int /*iovcnt*/,
                            ACE_Time_Value *timeout)
{
  return handler_->get_socket ()->recv (iov,this->peer_addr_,0,timeout);
}


//------------------------------------------------------------
// TAO_AV_UDP_Acceptor
//------------------------------------------------------------

TAO_AV_UDP_Acceptor::TAO_AV_UDP_Acceptor (void)
  : address_ (0),
    control_inet_address_ (0)
{
}

TAO_AV_UDP_Acceptor::~TAO_AV_UDP_Acceptor (void)
{
  if (this->flow_component_ == TAO_AV_Core::TAO_AV_CONTROL)
    delete this->entry_->control_handler ();

  delete this->address_;
  delete this->control_inet_address_;
}

int
TAO_AV_UDP_Acceptor::activate_svc_handler (TAO_AV_Flow_Handler *handler)
{
  ACE_Event_Handler *event_handler = handler->event_handler ();
  int result = this->av_core_->reactor ()->register_handler (event_handler,
                                                             ACE_Event_Handler::READ_MASK);

  if (this->flow_component_ == TAO_AV_Core::TAO_AV_CONTROL)
    handler->schedule_timer ();

 return result;
}

int
TAO_AV_UDP_Acceptor::open (TAO_Base_StreamEndPoint *endpoint,
                           TAO_AV_Core *av_core,
                           TAO_FlowSpec_Entry *entry,
                           TAO_AV_Flow_Protocol_Factory *factory,
                           TAO_AV_Core::Flow_Component flow_comp)
{
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"TAO_AV_UDP_Acceptor::open\n"));
  this->av_core_ = av_core;
  this->endpoint_ = endpoint;
  this->entry_ = entry;
  this->flow_component_ = flow_comp;