udp.cpp

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

    this->flow_protocol_factory_->make_protocol_object (this->entry_,
                                                        this->endpoint_,
                                                        flow_handler,
                                                        flow_handler->transport ());

  flow_handler->protocol_object (object);

  if (flow_component == TAO_AV_Core::TAO_AV_DATA)
    {
      this->endpoint_->set_flow_handler (this->flowname_.c_str (),
                                         flow_handler);
      this->entry_->protocol_object (object);
      entry->set_local_addr (local_addr);
      entry->handler (flow_handler);
      transport = flow_handler->transport ();
    }
  else
    {
      this->endpoint_->set_control_flow_handler (this->flowname_.c_str (),
                                                 flow_handler);
      this->entry_->control_protocol_object (object);
      entry->set_local_control_addr (local_addr);
      entry->control_handler (flow_handler);
      transport = flow_handler->transport ();
    }

  char buf[BUFSIZ];
  local_addr->addr_to_string (buf,BUFSIZ);

  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,"TAO_AV_UDP_CONNECTOR::connect:%s \n",buf));

  // call activate svc handler.
  return this->activate_svc_handler (flow_handler);
}

int
TAO_AV_UDP_Connector::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_Connector::close (void)
{
  return 0;
}

//------------------------------------------------------------
// TAO_AV_UDP_Connection_Setup
//------------------------------------------------------------

int
TAO_AV_UDP_Connection_Setup::setup (TAO_AV_Flow_Handler *&flow_handler,
                                    ACE_INET_Addr *inet_addr,
                                    ACE_INET_Addr *&local_addr,
                                    int is_multicast,
                                    ConnectionType ct)
{
  int result;

  if (is_multicast)
    {
      TAO_AV_UDP_MCast_Flow_Handler *handler;
      ACE_NEW_RETURN (handler,
                      TAO_AV_UDP_MCast_Flow_Handler,
                      -1);

      flow_handler = handler;

      result = handler->get_mcast_socket ()->subscribe (*inet_addr);
      if (result < 0)
        ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_UDP_MCast_connector::open failed\n"),-1);

      // Now disable Multicast loopback.
      // @@Should we make this a policy?
#if defined (ACE_HAS_IP_MULTICAST)
      if (handler->get_mcast_socket ()->set_option (IP_MULTICAST_LOOP,
                                                    0) < 0)
        if (TAO_debug_level > 0)
          ACE_DEBUG ((LM_DEBUG,"TAO_AV_UDP_MCast_Acceptor::multicast loop disable failed\n"));
      // @@ This should also be policies.
#endif /*ACE_HAS_IP_MULTICAST*/

      int bufsize = 80 * 1024;
      if (handler->get_mcast_socket ()->ACE_SOCK::set_option (SOL_SOCKET,
                                                              SO_RCVBUF,
                                                              (char *)&bufsize,
                                                              sizeof(bufsize)) < 0)
        {
          bufsize = 32 * 1024;
          if (handler->get_mcast_socket ()->ACE_SOCK::set_option (SOL_SOCKET,
                                                                  SO_RCVBUF,
                                                                  (char *)&bufsize,
                                                                  sizeof(bufsize)) < 0)
            perror("SO_RCVBUF");
        }
      ACE_NEW_RETURN (local_addr,
                      ACE_INET_Addr ("0"),
                      -1);

      if (ct == ACCEPTOR)
        {
          result = handler->get_mcast_socket ()->get_local_addr (*local_addr);
          if (result < 0)
            ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_Dgram_Connector::open: get_local_addr failed\n"),result);

          local_addr->set (local_addr->get_port_number (),
                           local_addr->get_host_name ());
	  handler->set_peer_addr (local_addr);
        }
    }
  else
    {
      if (local_addr == 0)
	ACE_NEW_RETURN (local_addr,
			ACE_INET_Addr ("0"),
			-1);

      TAO_AV_UDP_Flow_Handler *handler;
      ACE_NEW_RETURN (handler,
                      TAO_AV_UDP_Flow_Handler,
                      -1);

      flow_handler = handler;

      if (ct == ACCEPTOR)
        result = handler->open (*inet_addr);
      else
	result = handler->open (*local_addr);
      if (result < 0)
        ACE_ERROR_RETURN ((LM_ERROR,"handler::open failed\n"),-1);

      // set the socket buffer sizes to 64k.
      int sndbufsize = ACE_DEFAULT_MAX_SOCKET_BUFSIZ;
      int rcvbufsize = ACE_DEFAULT_MAX_SOCKET_BUFSIZ;

      if (handler->get_socket ()->set_option (SOL_SOCKET,
                                              SO_SNDBUF,
                                              (void *) &sndbufsize,
                                              sizeof (sndbufsize)) == -1
          && errno != ENOTSUP)
        return 0;
      else if (handler->get_socket ()->set_option (SOL_SOCKET,
                                                   SO_RCVBUF,
                                                   (void *) &rcvbufsize,
                                                   sizeof (rcvbufsize)) == -1
               && errno != ENOTSUP)
        return 0;

      if (ct == CONNECTOR)
	handler->set_remote_address  (inet_addr);

      result = handler->get_socket ()->get_local_addr (*local_addr);

      local_addr->set (local_addr->get_port_number (),
		       local_addr->get_host_name ());

      char buf [BUFSIZ];
      local_addr->addr_to_string (buf, BUFSIZ);

      if (result < 0)
        ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_Dgram_Connector::open: get_local_addr failed\n"),result);
    }

  return 1;
}

//------------------------------------------------------------
// TAO_AV_UDP_Protocol_Factory
//------------------------------------------------------------

TAO_AV_UDP_Factory::TAO_AV_UDP_Factory (void)
{
}

TAO_AV_UDP_Factory::~TAO_AV_UDP_Factory (void)
{
}

int
TAO_AV_UDP_Factory::match_protocol (const char *protocol_string)
{
  if (ACE_OS::strcasecmp (protocol_string,"UDP") == 0)
    return 1;
  if (ACE_OS::strcasecmp (protocol_string,"RTP/UDP") == 0)
    return 1;
  return 0;
}

TAO_AV_Acceptor*
TAO_AV_UDP_Factory::make_acceptor (void)
{
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"TAO_AV_UDP_Factory::make_acceptor\n"));
  TAO_AV_Acceptor *acceptor = 0;
  ACE_NEW_RETURN (acceptor,
                  TAO_AV_UDP_Acceptor,
                  0);
  return acceptor;
}

TAO_AV_Connector*
TAO_AV_UDP_Factory::make_connector (void)
{
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"TAO_AV_UDP_Factory::make_connector\n"));
  TAO_AV_Connector *connector = 0;
  ACE_NEW_RETURN (connector,
                  TAO_AV_UDP_Connector,
                  0);
  return connector;
}

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

//------------------------------------------------------------
// TAO_AV_UDP_Object
//------------------------------------------------------------

int
TAO_AV_UDP_Object::handle_input (void)
{
  int n = this->transport_->recv (this->frame_.rd_ptr (),
                                  this->frame_.size ());
  if (n == -1)
    ACE_ERROR_RETURN ((LM_ERROR,"(%N,%l) TAO_AV_UDP_Flow_Handler::handle_input recv failed: errno: %m\n"),-1);

  this->frame_.wr_ptr (this->frame_.rd_ptr () + n);

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

int
TAO_AV_UDP_Object::send_frame (ACE_Message_Block *frame,
                               TAO_AV_frame_info * /*frame_info*/)
{
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"TAO_AV_UDP_Object::send_frame\n"));
  int result = this->transport_->send (frame);
  if (result < 0)
    return result;
  return 0;
}

int
TAO_AV_UDP_Object::send_frame (const iovec *iov,
                               int iovcnt,
                               TAO_AV_frame_info * /*frame_info*/)
{
  int result = this->transport_->send (iov,iovcnt);
  if (result < 0)
    return result;
  return 0;
}

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

TAO_AV_UDP_Object::TAO_AV_UDP_Object (TAO_AV_Callback *callback,
                                      TAO_AV_Transport *transport)
  :TAO_AV_Protocol_Object (callback,transport)
{
  this->frame_.size (this->transport_->mtu ());
}

TAO_AV_UDP_Object::~TAO_AV_UDP_Object (void)
{
  //no-op
}

int
TAO_AV_UDP_Object::destroy (void)
{
  this->callback_->handle_destroy ();
  delete this;

  return 0;
}


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

TAO_AV_UDP_Flow_Factory::~TAO_AV_UDP_Flow_Factory (void)
{
}

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

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

TAO_AV_Protocol_Object*
TAO_AV_UDP_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_UDP_Object *object = 0;
  ACE_NEW_RETURN (object,
                  TAO_AV_UDP_Object (callback,
                                     transport),
                  0);
  callback->open (object,
                  handler);
  endpoint->set_protocol_object (entry->flowname (),
                                 object);
  return object;
}

ACE_FACTORY_DEFINE (TAO_AV, TAO_AV_UDP_Flow_Factory)
ACE_STATIC_SVC_DEFINE (TAO_AV_UDP_Flow_Factory,
                       ACE_TEXT ("UDP_Flow_Factory"),
                       ACE_SVC_OBJ_T,
                       &ACE_SVC_NAME (TAO_AV_UDP_Flow_Factory),
                       ACE_Service_Type::DELETE_THIS |
                       ACE_Service_Type::DELETE_OBJ,
                       0)

ACE_FACTORY_DEFINE (TAO_AV, TAO_AV_UDP_Factory)
ACE_STATIC_SVC_DEFINE (TAO_AV_UDP_Factory,
                       ACE_TEXT ("UDP_Factory"),
                       ACE_SVC_OBJ_T,
                       &ACE_SVC_NAME (TAO_AV_UDP_Factory),
                       ACE_Service_Type::DELETE_THIS |
                       ACE_Service_Type::DELETE_OBJ,
                       0)