ecg_mcast_gateway.cpp

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

    ? this->address_server_arg_.c_str () : 0;

  if (this->address_server_type_ == ECG_ADDRESS_SERVER_BASIC)
    {
      TAO_EC_Servant_Var<TAO_ECG_Simple_Address_Server> impl =
        TAO_ECG_Simple_Address_Server::create ();
      if (!impl.in ())
        return 0;

      if (impl->init (address_server_arg) == -1)
        {
          return 0;
        }
      return impl._retn ();
    }

  else if (this->address_server_type_ == ECG_ADDRESS_SERVER_SOURCE)
    {
      TAO_EC_Servant_Var<TAO_ECG_Complex_Address_Server> impl =
        TAO_ECG_Complex_Address_Server::create (1);
      if (!impl.in ())
        return 0;

      if (impl->init (address_server_arg) == -1)
        {
          return 0;
        }
      return impl._retn ();
    }

  else if (this->address_server_type_ == ECG_ADDRESS_SERVER_TYPE)
    {
      TAO_EC_Servant_Var<TAO_ECG_Complex_Address_Server> impl =
        TAO_ECG_Complex_Address_Server::create (0);
      if (!impl.in ())
        return 0;

      if (impl->init (address_server_arg) == -1)
        {
          return 0;
        }
      return impl._retn ();
    }

  else
    {
      ACE_ERROR ((LM_ERROR,
                  "Cannot create address server: "
                  "unknown address server type specified."));
      return 0;
    }
}

TAO_ECG_Refcounted_Handler
TAO_ECG_Mcast_Gateway::init_handler (TAO_ECG_Dgram_Handler *receiver,
                                     RtecEventChannelAdmin::EventChannel_ptr ec,
                                     ACE_Reactor *reactor
                                     ACE_ENV_ARG_DECL)
{
  TAO_ECG_Refcounted_Handler handler;

  const char * nic =
    (this->nic_.length ()) ? this->nic_.c_str () : 0;
  const char * address_server_arg =
    (this->address_server_arg_.length ())
    ? this->address_server_arg_.c_str () : 0;

  if (this->handler_type_ == ECG_HANDLER_BASIC)
    {
      TAO_ECG_Simple_Mcast_EH * h = 0;
      ACE_NEW_RETURN (h,
                      TAO_ECG_Simple_Mcast_EH (receiver),
                      handler);
      handler = h;

      h->reactor (reactor);
      if (h->open (address_server_arg, nic) != 0)
        return TAO_ECG_Refcounted_Handler ();
    }

  else if (this->handler_type_ == ECG_HANDLER_COMPLEX)
    {
      TAO_ECG_Mcast_EH * h = 0;
      ACE_NEW_RETURN (h,
                      TAO_ECG_Mcast_EH (receiver, nic),
                      handler);
      handler = h;

      h->reactor (reactor);

      h->open (ec ACE_ENV_ARG_PARAMETER);
      ACE_CHECK_RETURN (TAO_ECG_Refcounted_Handler ());
    }

  else if (this->handler_type_ == ECG_HANDLER_UDP)
    {
      TAO_ECG_UDP_EH * h = 0;
      ACE_NEW_RETURN (h,
                      TAO_ECG_UDP_EH (receiver),
                      handler);
      handler = h;
      h->reactor (reactor);

      ACE_INET_Addr ipaddr;
      if (ipaddr.set (address_server_arg) != 0)
        {
          ACE_ERROR ((LM_ERROR,
                      "ERROR using address server argument "
                      "in ACE_INET_Addr.set ()."));
          return TAO_ECG_Refcounted_Handler ();
        }
      if (h->open (ipaddr) != 0)
        return TAO_ECG_Refcounted_Handler ();
    }

  else
    {
      ACE_ERROR ((LM_ERROR,
                  "Cannot create handler: unknown "
                  "handler type specified."));
      return handler;
    }

  return handler;
}

TAO_EC_Servant_Var<TAO_ECG_UDP_Sender>
TAO_ECG_Mcast_Gateway::init_sender (
                               RtecEventChannelAdmin::EventChannel_ptr ec,
                               RtecUDPAdmin::AddrServer_ptr address_server,
                               TAO_ECG_Refcounted_Endpoint endpoint_rptr
                               ACE_ENV_ARG_DECL)
{
  TAO_EC_Servant_Var<TAO_ECG_UDP_Sender>
    sender (TAO_ECG_UDP_Sender::create ());
  if (!sender.in ())
    return sender;

  sender->init (ec,
                address_server,
                endpoint_rptr
                ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (TAO_EC_Servant_Var<TAO_ECG_UDP_Sender> ());

  TAO_EC_Auto_Command<UDP_Sender_Shutdown> sender_shutdown;
  sender_shutdown.set_command (UDP_Sender_Shutdown (sender));

  if (this->consumer_qos_.dependencies.length () > 0)
    {
      // Client supplied consumer qos.  Use it.
      this->consumer_qos_.is_gateway = 1;
      sender->connect (this->consumer_qos_ ACE_ENV_ARG_PARAMETER);
      ACE_CHECK_RETURN (TAO_EC_Servant_Var<TAO_ECG_UDP_Sender> ());
    }
  else
    {
      // Client did not specify anything - subscribe to all events.
      ACE_ConsumerQOS_Factory consumer_qos_factory;
      consumer_qos_factory.start_disjunction_group (1);
      consumer_qos_factory.insert (ACE_ES_EVENT_SOURCE_ANY,
                                   ACE_ES_EVENT_ANY,
                                   0);
      RtecEventChannelAdmin::ConsumerQOS & qos =
        ACE_const_cast (RtecEventChannelAdmin::ConsumerQOS &,
                        consumer_qos_factory.get_ConsumerQOS ());
      qos.is_gateway = 1;

      sender->connect (qos ACE_ENV_ARG_PARAMETER);
      ACE_CHECK_RETURN (TAO_EC_Servant_Var<TAO_ECG_UDP_Sender> ());
    }

  sender_shutdown.disallow_command ();
  return sender;
}

TAO_EC_Servant_Var<TAO_ECG_UDP_Receiver>
TAO_ECG_Mcast_Gateway::init_receiver (
                               RtecEventChannelAdmin::EventChannel_ptr ec,
                               RtecUDPAdmin::AddrServer_ptr address_server,
                               TAO_ECG_Refcounted_Endpoint endpoint_rptr
                               ACE_ENV_ARG_DECL)
{
  TAO_EC_Servant_Var<TAO_ECG_UDP_Receiver>
    receiver (TAO_ECG_UDP_Receiver::create ());
  if (!receiver.in ())
    return receiver;

  receiver->init (ec,
                  endpoint_rptr,
                  address_server
                  ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (TAO_EC_Servant_Var<TAO_ECG_UDP_Receiver> ());

  TAO_EC_Auto_Command<UDP_Receiver_Shutdown> receiver_shutdown;
  receiver_shutdown.set_command (UDP_Receiver_Shutdown (receiver));

  ACE_SupplierQOS_Factory supplier_qos_factory;
  supplier_qos_factory.insert (ACE_ES_EVENT_SOURCE_ANY,
                               ACE_ES_EVENT_ANY,
                               0, 1);
  RtecEventChannelAdmin::SupplierQOS & qos =
    ACE_const_cast (RtecEventChannelAdmin::SupplierQOS &,
                    supplier_qos_factory.get_SupplierQOS ());
  qos.is_gateway = 1;

  receiver->connect (qos ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (TAO_EC_Servant_Var<TAO_ECG_UDP_Receiver> ());

  receiver_shutdown.disallow_command ();
  return receiver;
}

void
TAO_ECG_Mcast_Gateway::verify_args (CORBA::ORB_ptr orb,
                                    RtecEventChannelAdmin::EventChannel_ptr ec
                                    ACE_ENV_ARG_DECL)
{
  if (CORBA::is_nil (ec))
    {
      ACE_ERROR ((LM_ERROR,
                  "Nil event channel argument passed to "
                  "TAO_ECG_Mcast_Gateway::run()."));
      ACE_THROW (CORBA::INTERNAL ());
    }
  if (CORBA::is_nil (orb))
    {
      ACE_ERROR ((LM_ERROR,
                  "Nil orb argument passed to "
                  "TAO_ECG_Mcast_Gateway::run()."));
      ACE_THROW (CORBA::INTERNAL ());
    }
}

void
TAO_ECG_Mcast_Gateway::run (CORBA::ORB_ptr orb,
                            RtecEventChannelAdmin::EventChannel_ptr ec
                            ACE_ENV_ARG_DECL)
{
  // Verify args.
  this->verify_args (orb, ec ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  // Auto-cleanup objects.
  TAO_EC_Object_Deactivator address_server_deactivator;
  TAO_EC_Auto_Command<UDP_Sender_Shutdown> sender_shutdown;
  TAO_EC_Auto_Command<UDP_Receiver_Shutdown> receiver_shutdown;

  // Set up address server.
  PortableServer::ServantBase_var address_server_servant =
    this->init_address_server ();
  if (!address_server_servant.in ())
    {
      ACE_DEBUG ((LM_ERROR,
                  "Unable to create address server."));
      ACE_THROW (CORBA::INTERNAL ());
    }

  RtecUDPAdmin::AddrServer_var address_server;

  PortableServer::POA_var poa =
    address_server_servant->_default_POA (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  activate (address_server,
            poa.in (),
            address_server_servant.in (),
            address_server_deactivator
            ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  TAO_ECG_Refcounted_Endpoint endpoint_rptr;
  TAO_EC_Servant_Var<TAO_ECG_UDP_Sender> sender;

  // Set up event sender.
  if (this->service_type_ == ECG_MCAST_SENDER
      || this->service_type_ == ECG_MCAST_TWO_WAY)
    {
      endpoint_rptr = this->init_endpoint ();
      if (endpoint_rptr.get () == 0)
        {
          ACE_THROW (CORBA::INTERNAL ());
        }

      sender = this->init_sender (ec,
                                  address_server.in (),
                                  endpoint_rptr
                                  ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
      if (!sender.in ())
        {
          ACE_THROW (CORBA::INTERNAL ());
        }

      sender_shutdown.set_command (UDP_Sender_Shutdown (sender));
    }

  // Set up event receiver.
  TAO_EC_Servant_Var<TAO_ECG_UDP_Receiver> receiver;
  if (this->service_type_ == ECG_MCAST_RECEIVER
      || this->service_type_ == ECG_MCAST_TWO_WAY)
    {
      ACE_Reactor *reactor = orb->orb_core ()->reactor ();

      receiver = this->init_receiver (ec,
                                      address_server.in (),
                                      endpoint_rptr
                                      ACE_ENV_ARG_PARAMETER);
      ACE_CHECK;
      if (!receiver.in ())
        {
          ACE_THROW (CORBA::INTERNAL ());
        }

      receiver_shutdown.set_command (UDP_Receiver_Shutdown (receiver));

      TAO_ECG_Refcounted_Handler
        handler_rptr (this->init_handler (receiver.in (),
                                          ec,
                                          reactor
                                          ACE_ENV_ARG_PARAMETER));
      ACE_CHECK;
      if (handler_rptr.get () == 0)
        {
          ACE_THROW (CORBA::INTERNAL ());
        }
      receiver->set_handler_shutdown (handler_rptr);
    }

  // Everything went ok - disable auto-cleanup.
  address_server_deactivator.disallow_deactivation ();
  receiver_shutdown.disallow_command ();
  sender_shutdown.disallow_command ();
}

// ****************************************************************

ACE_STATIC_SVC_DEFINE (TAO_ECG_Mcast_Gateway,
                       ACE_TEXT ("ECG_Mcast_Gateway"),
                       ACE_SVC_OBJ_T,
                       &ACE_SVC_NAME (TAO_ECG_Mcast_Gateway),
                       ACE_Service_Type::DELETE_THIS | ACE_Service_Type::DELETE_OBJ,
                       0)
ACE_FACTORY_DEFINE (TAO_RTEvent, TAO_ECG_Mcast_Gateway)

// ****************************************************************


#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)

template class ACE_Dynamic_Service<TAO_ECG_Mcast_Gateway>;
template void activate<>(TAO_Objref_Var_T<RtecUDPAdmin::AddrServer, RtecUDPAdmin::tao_AddrServer_life>&, PortableServer::POA*, TAO_ServantBase*, TAO_EC_Object_Deactivator& ACE_ENV_ARG_DECL);

#elif defined(ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)

#pragma instantiate ACE_Dynamic_Service<TAO_ECG_Mcast_Gateway>
#pragma instantiate activate<>(TAO_Objref_Var_T<RtecUDPAdmin::AddrServer, RtecUDPAdmin::tao_AddrServer_life>&, PortableServer::POA*, TAO_ServantBase*, TAO_EC_Object_Deactivator& ACE_ENV_ARG_DECL)

#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */