TAO_ECG_Mcast_Gateway Class Reference

Implement the builder for setting up Event Channel multicast gateway. NOT THREAD-SAFE. More...

#include <ECG_Mcast_Gateway.h>

Inheritance diagram for TAO_ECG_Mcast_Gateway:

[legend]Collaboration diagram for TAO_ECG_Mcast_Gateway:

[legend]List of all members.

Public Types
	ECG_MCAST_SENDER
	ECG_MCAST_RECEIVER
	ECG_MCAST_TWO_WAY
	ECG_ADDRESS_SERVER_BASIC
	ECG_ADDRESS_SERVER_SOURCE
	ECG_ADDRESS_SERVER_TYPE
	ECG_HANDLER_BASIC
	ECG_HANDLER_COMPLEX
	ECG_HANDLER_UDP
enum	Service_Type { ECG_MCAST_SENDER, ECG_MCAST_RECEIVER, ECG_MCAST_TWO_WAY }
	Values for some configuration parameters to init (). More...
enum	Address_Server_Type { ECG_ADDRESS_SERVER_BASIC, ECG_ADDRESS_SERVER_SOURCE, ECG_ADDRESS_SERVER_TYPE }
	Values for some configuration parameters to init (). More...
enum	Handler_Type { ECG_HANDLER_BASIC, ECG_HANDLER_COMPLEX, ECG_HANDLER_UDP }
	Values for some configuration parameters to init (). More...
Public Member Functions
	TAO_ECG_Mcast_Gateway (void)
	Constructor.
int	init (const char *address_server_arg, const Attributes &attributes=Attributes())
int	init (const RtecEventChannelAdmin::ConsumerQOS &consumer_qos, const char *address_server_arg, const Attributes &attributes=Attributes())
void	run (CORBA::ORB_ptr orb, RtecEventChannelAdmin::EventChannel_ptr ec)
virtual int	init (int argc, ACE_TCHAR *argv[])
	The Service_Object entry points.
virtual int	fini (void)
	The Service_Object entry points.
Static Public Member Functions
static int	init_svcs (void)
Private Member Functions
void	verify_args (CORBA::ORB_ptr orb, RtecEventChannelAdmin::EventChannel_ptr ec)
	Check that arguments to run() are not nil.
int	validate_configuration (void)
	Verifies configuration values specified through init() make sense.
PortableServer::ServantBase *	init_address_server (void)
	Allocate and initialize appropriate objects.
TAO_EC_Servant_Var< TAO_ECG_UDP_Sender >	init_sender (RtecEventChannelAdmin::EventChannel_ptr ec, RtecUDPAdmin::AddrServer_ptr address_server, TAO_ECG_Refcounted_Endpoint endpoint_rptr)
	Allocate and initialize appropriate objects.
TAO_EC_Servant_Var< TAO_ECG_UDP_Receiver >	init_receiver (RtecEventChannelAdmin::EventChannel_ptr ec, RtecUDPAdmin::AddrServer_ptr address_server, TAO_ECG_Refcounted_Endpoint endpoint_rptr)
	Allocate and initialize appropriate objects.
TAO_ECG_Refcounted_Endpoint	init_endpoint (void)
	Allocate and initialize appropriate objects.
TAO_ECG_Refcounted_Handler	init_handler (TAO_ECG_Dgram_Handler *recv, RtecEventChannelAdmin::EventChannel_ptr ec, ACE_Reactor *reactor)
	Allocate and initialize appropriate objects.
Private Attributes
Service_Type	service_type_
	Flags controlling configuration.
Handler_Type	handler_type_
	Flags controlling configuration.
Address_Server_Type	address_server_type_
	Flags controlling configuration.
ACE_CString	address_server_arg_
	Flags controlling configuration.
u_char	ttl_value_
	Flags controlling configuration.
ACE_CString	nic_
	Flags controlling configuration.
int	ip_multicast_loop_
	Flags controlling configuration.
int	non_blocking_
	Flags controlling configuration.
RtecEventChannelAdmin::ConsumerQOS	consumer_qos_
	Flags controlling configuration.
Classes
struct	Attributes
	Helper class to initialize a TAO_ECG_Mcast_Gateway. More...

---

Detailed Description

Implement the builder for setting up Event Channel multicast gateway. NOT THREAD-SAFE.

This class simplifies creation of federated Event Channels by presenting a simple unified interface for creating and configuring all components needed to federate an Event Channel. Configuration options are described below.

NOTE: This class does not own any of the components it creates and its lifetime is independent of theirs. This class acts purely as a wrapper facade for creating and wiring appropriate components together.

Todo:

This class is an ACE_Service_Object, but the only reason for it is the need for easy configuration using files. Since ACE_Service_Object provides much more than that, we should look into replacing it with a more lightweight utility that would serve our needs.

CONFIGURATION OPTIONS There are two ways to use this class: 1) Use service config file to specify configuration options (which are described below), and use service configurator to obtain a an instance of configured TAO_ECG_Mcast_Gateway in your program. (See TAO/orbsvcs/tests/Event/Mcast/Common and TAO/orbsvcs/tests/Event/Mcast/Simple for an example.)

Service config file options:

-ECGService <service> Valid values: sender, receiver, two_way Specifies whether this gateway should act as a multicast sender of the events or multicast receiver, or both.

-ECGAddressServer <server_type> Valid values: basic, source, type Specifies what implementation of the address server should be used by the gateway. basic - the same multicast address is returned for all event headers. source - multicast addresses are returned based on the event source, according to the mapping provided at initialization. type - multicast addresses are returned based on the event type, according to the mapping provided at initialization.

-ECGAddressServerArg <arg> Valid value: arg is a string, whose format requirements are specific to the implementation of address server used. Arg is not interpreted by the gateway, but simply passed to the address server specified by -ECGAddressServer during initialization. THIS OPTION MUST ALWAYS BE SPECIFIED BY THE USER (there is no default value for it)

-ECGHandler <handler_type> Valid values: basic, complex, udp Specifies what implementation of event handler should be used if gateway is acting as events receiver. basic - a simple event handler listening on a single mcast address. complex - event handler listening on multiple mcast addresses based on events of interest to consumers. udp - similar to basic handler, except listens on udp address as opposed to a multicast group.

-ECGTTL <ttl> Valid values: a number > 0 IP_Multicast time to live value that should be set on a sending socket. This option matters only if the gateway is acting as a sender of mcast messages.

-ECGNIC <nic> Valid values: name of the network interface This interface is used for sending and/or receiving multicast messages.

-ECGNonBlocking Boolean flag to configure if the socket is in blocking or non-blocking code. The default is non-blocking. NOTE: Certain device drivers block the process if the physical link fails.

2) Create an instance of TAO_ECG_Mcast_Gateway in your code, on the stack or dynamically, and use init () method to configure it. No configuration files involved. See service config options above for the description of configurable options, and init() method below for how to specify them.

Default configuration values (for either use case) can be found in ECG_Defaults.h

Definition at line 128 of file ECG_Mcast_Gateway.h.

---

Member Enumeration Documentation

enum TAO_ECG_Mcast_Gateway::Address_Server_Type

Values for some configuration parameters to init ().

Enumerator:

ECG_ADDRESS_SERVER_BASIC
ECG_ADDRESS_SERVER_SOURCE
ECG_ADDRESS_SERVER_TYPE

Definition at line 152 of file ECG_Mcast_Gateway.h.

                           {ECG_ADDRESS_SERVER_BASIC,
                            ECG_ADDRESS_SERVER_SOURCE,
                            ECG_ADDRESS_SERVER_TYPE};

enum TAO_ECG_Mcast_Gateway::Handler_Type

Values for some configuration parameters to init ().

Enumerator:

ECG_HANDLER_BASIC
ECG_HANDLER_COMPLEX
ECG_HANDLER_UDP

Definition at line 156 of file ECG_Mcast_Gateway.h.

                    {ECG_HANDLER_BASIC,
                     ECG_HANDLER_COMPLEX,
                     ECG_HANDLER_UDP};

enum TAO_ECG_Mcast_Gateway::Service_Type

Values for some configuration parameters to init ().

Enumerator:

ECG_MCAST_SENDER
ECG_MCAST_RECEIVER
ECG_MCAST_TWO_WAY

Definition at line 148 of file ECG_Mcast_Gateway.h.

                    {ECG_MCAST_SENDER,
                     ECG_MCAST_RECEIVER,
                     ECG_MCAST_TWO_WAY};

---

Constructor & Destructor Documentation

TAO_ECG_Mcast_Gateway::TAO_ECG_Mcast_Gateway

(

void

)

Constructor.

---

Member Function Documentation

int TAO_ECG_Mcast_Gateway::fini

(

void

)

[virtual]

The Service_Object entry points.

Reimplemented from ACE_Shared_Object.

Definition at line 44 of file ECG_Mcast_Gateway.cpp.

{
  return 0;
}

int TAO_ECG_Mcast_Gateway::init	(	const RtecEventChannelAdmin::ConsumerQOS &	consumer_qos,
		const char *	address_server_arg,
		const Attributes &	attributes = Attributes()
	)

Same as the method above, but also gives the client an opportunity to specify consumer qos, i.e., which EC traffic should get multicasted.

Definition at line 233 of file ECG_Mcast_Gateway.cpp.

References consumer_qos_, and init().

{
  this->consumer_qos_ = consumer_qos;
  return this->init (address_server_arg,
                     attributes);
}

int TAO_ECG_Mcast_Gateway::init	(	const char *	address_server_arg,
		const Attributes &	attributes = Attributes()
	)

Configure TAO_ECG_Mcast_Gateway programatically. This method should be used when NOT using service configurator to obtain/configure TAO_ECG_Mcast_Gateway. See class documentation above for more info.

Definition at line 216 of file ECG_Mcast_Gateway.cpp.

References address_server_arg_, TAO_ECG_Mcast_Gateway::Attributes::address_server_type, address_server_type_, ACE_String_Base< CHAR >::c_str(), TAO_ECG_Mcast_Gateway::Attributes::handler_type, handler_type_, TAO_ECG_Mcast_Gateway::Attributes::ip_multicast_loop, ip_multicast_loop_, TAO_ECG_Mcast_Gateway::Attributes::nic, nic_, TAO_ECG_Mcast_Gateway::Attributes::non_blocking, non_blocking_, TAO_ECG_Mcast_Gateway::Attributes::service_type, service_type_, ACE_String_Base< CHAR >::set(), TAO_ECG_Mcast_Gateway::Attributes::ttl_value, ttl_value_, and validate_configuration().

{
  this->address_server_arg_.set (address_server_arg);

  this->address_server_type_ = attr.address_server_type;
  this->handler_type_ = attr.handler_type;
  this->service_type_ = attr.service_type;
  this->ttl_value_ = attr.ttl_value;
  this->nic_.set (attr.nic.c_str ());
  this->ip_multicast_loop_ = attr.ip_multicast_loop;
  this->non_blocking_ = attr.non_blocking;

  return this->validate_configuration ();
}

int TAO_ECG_Mcast_Gateway::init	(	int	argc,
		ACE_TCHAR *	argv[]
	)			[virtual]

The Service_Object entry points.

Reimplemented from ACE_Shared_Object.

Definition at line 50 of file ECG_Mcast_Gateway.cpp.

References ACE_DEBUG, ACE_ERROR, ACE_TEXT(), address_server_type_, ACE_OS::atoi(), ACE_Arg_Shifter_T< CHAR_TYPE >::consume_arg(), ECG_ADDRESS_SERVER_BASIC, ECG_ADDRESS_SERVER_SOURCE, ECG_ADDRESS_SERVER_TYPE, ECG_HANDLER_BASIC, ECG_HANDLER_COMPLEX, ECG_HANDLER_UDP, ECG_MCAST_RECEIVER, ECG_MCAST_SENDER, ECG_MCAST_TWO_WAY, ACE_Arg_Shifter_T< CHAR_TYPE >::get_current(), handler_type_, ACE_Arg_Shifter_T< CHAR_TYPE >::ignore_arg(), ACE_Arg_Shifter_T< CHAR_TYPE >::is_anything_left(), ACE_Arg_Shifter_T< CHAR_TYPE >::is_parameter_next(), LM_ERROR, LM_WARNING, service_type_, ACE_OS::strcasecmp(), ACE_OS::strtoul(), and ttl_value_.

Referenced by init().

{
  int result = 0;

  ACE_Arg_Shifter arg_shifter (argc, argv);

  while (arg_shifter.is_anything_left ())
    {
      const ACE_TCHAR *arg = arg_shifter.get_current ();

      if (ACE_OS::strcasecmp (arg, ACE_TEXT ("-ECGService")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, ACE_TEXT ("receiver")) == 0)
                this->service_type_ = ECG_MCAST_RECEIVER;
              else if (ACE_OS::strcasecmp (opt, ACE_TEXT ("sender")) == 0)
                this->service_type_ = ECG_MCAST_SENDER;
              else if (ACE_OS::strcasecmp (opt, ACE_TEXT ("two_way")) == 0)
                this->service_type_ = ECG_MCAST_TWO_WAY;
              else
                {
                  ACE_ERROR ((LM_ERROR,
                             ACE_TEXT ("Unsupported <-ECGService> option ")
                             ACE_TEXT ("value: <%s>. Ignoring this option ")
                             ACE_TEXT ("- using defaults instead.\n"),
                             opt));
                  result = -1;
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_TEXT ("-ECGAddressServer")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, ACE_TEXT ("basic")) == 0)
                this->address_server_type_ = ECG_ADDRESS_SERVER_BASIC;
              else if (ACE_OS::strcasecmp (opt, ACE_TEXT ("source")) == 0)
                this->address_server_type_ = ECG_ADDRESS_SERVER_SOURCE;
              else if (ACE_OS::strcasecmp (opt, ACE_TEXT ("type")) == 0)
                this->address_server_type_ = ECG_ADDRESS_SERVER_TYPE;
              else
                {
                  ACE_ERROR ((LM_ERROR,
                              ACE_TEXT ("Unsupported <-ECGAddressServer> ")
                              ACE_TEXT ("option value: <%s>. Ignoring this ")
                              ACE_TEXT ("option - using defaults instead.\n"),
                              opt));
                  result = -1;
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_TEXT ("-ECGAddressServerArg")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              this->address_server_arg_.set (arg_shifter.get_current ());
              arg_shifter.consume_arg ();
            }
        }


      else if (ACE_OS::strcasecmp (arg, ACE_TEXT ("-ECGHandler")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, ACE_TEXT ("basic")) == 0)
                this->handler_type_ = ECG_HANDLER_BASIC;
              else if (ACE_OS::strcasecmp (opt, ACE_TEXT ("complex")) == 0)
                this->handler_type_ = ECG_HANDLER_COMPLEX;
              else if (ACE_OS::strcasecmp (opt, ACE_TEXT ("udp")) == 0)
                this->handler_type_ = ECG_HANDLER_UDP;
              else
                {
                  ACE_ERROR ((LM_ERROR,
                              ACE_TEXT ("Unsupported <-ECGHandler> ")
                              ACE_TEXT ("option value: <%s>. Ignoring this ")
                              ACE_TEXT ("option - using defaults instead.\n"),
                              opt));
                  result = -1;
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_TEXT ("-ECGTTL")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const ACE_TCHAR* opt = arg_shifter.get_current ();
              unsigned long tmp = ACE_OS::strtoul (opt, 0, 0) & 0xff;
              this->ttl_value_ = static_cast<u_char> (tmp);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_TEXT ("-ECGNIC")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              this->nic_.set (arg_shifter.get_current ());
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_TEXT ("-ECGIPMULTICASTLOOP")) == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              this->ip_multicast_loop_ =
                (ACE_OS::atoi(arg_shifter.get_current()) != 0);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, ACE_TEXT ("-ECGNONBLOCKING")) == 0)
      {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              this->non_blocking_ =
                (ACE_OS::atoi(arg_shifter.get_current()) != 0);
              arg_shifter.consume_arg ();
            }
        }

      else
        {
          arg_shifter.ignore_arg ();
          ACE_DEBUG ((LM_WARNING,
                      ACE_TEXT ("Ignoring <%s> option ")
                      ACE_TEXT ("during initialization.\n"),
                      arg));
          result = -1;
        }
    }

  if (this->validate_configuration () == -1)
    return -1;
  else
    return result;
}

PortableServer::ServantBase * TAO_ECG_Mcast_Gateway::init_address_server

(

void

)

[private]

Allocate and initialize appropriate objects.

Definition at line 359 of file ECG_Mcast_Gateway.cpp.

References TAO_EC_Servant_Var< T >::_retn(), ACE_ERROR, address_server_arg_, TAO_ECG_Complex_Address_Server::create(), TAO_ECG_Simple_Address_Server::create(), ECG_ADDRESS_SERVER_BASIC, ECG_ADDRESS_SERVER_SOURCE, ECG_ADDRESS_SERVER_TYPE, TAO_EC_Servant_Var< T >::in(), ACE_String_Base< CHAR >::length(), and LM_ERROR.

Referenced by run().

{
  const char * address_server_arg =
    (this->address_server_arg_.length ())
    ? 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.\n"));
      return 0;
    }
}

TAO_ECG_Refcounted_Endpoint TAO_ECG_Mcast_Gateway::init_endpoint

(

void

)

[private]

Allocate and initialize appropriate objects.

Definition at line 288 of file ECG_Mcast_Gateway.cpp.

References ACE_ERROR, ACE_NEW_NORETURN, ACE_NONBLOCK, ACE_IPC_SAP::enable(), IP_MULTICAST_LOOP, IP_MULTICAST_TTL, IPPROTO_IP, LM_ERROR, ACE_SOCK_Dgram::open(), ACE_Refcounted_Auto_Ptr< X, ACE_LOCK >::reset(), ACE_Addr::sap_any, and ACE_SOCK_Dgram::set_nic().

Referenced by run().

{
  TAO_ECG_UDP_Out_Endpoint* endpoint = 0;
  TAO_ECG_Refcounted_Endpoint refendpoint;

  // Try to allocate a new endpoint from the heap
  ACE_NEW_NORETURN (endpoint,
                    TAO_ECG_UDP_Out_Endpoint);

  if (endpoint != 0)
  {
    refendpoint.reset (endpoint);
  }
  else
  {
    return TAO_ECG_Refcounted_Endpoint ();
  }

  ACE_SOCK_Dgram& dgram = refendpoint->dgram ();

  if (dgram.open (ACE_Addr::sap_any) == -1)
    {
      ACE_ERROR ((LM_ERROR,
                             "Cannot open dgram "
                             "for sending mcast messages.\n"));
      return TAO_ECG_Refcounted_Endpoint ();
    }

  if (this->nic_.length () != 0)
    {
      dgram.set_nic (this->nic_.c_str ());
    }

  if (this->ttl_value_ > 0)
    {
      if (dgram.ACE_SOCK::set_option (IPPROTO_IP,
                                      IP_MULTICAST_TTL,
                                      &this->ttl_value_,
                                      sizeof (this->ttl_value_))
          == -1)
        {
          ACE_ERROR ((LM_ERROR,
                      "Error setting TTL option on dgram "
                      "for sending mcast messages.\n"));
          return TAO_ECG_Refcounted_Endpoint ();
        }
    }

  if (dgram.ACE_SOCK::set_option (IPPROTO_IP,
                                  IP_MULTICAST_LOOP,
                                  &this->ip_multicast_loop_,
                                  sizeof (this->ip_multicast_loop_)) == -1)
    {
      ACE_ERROR ((LM_ERROR,
                  "Error setting MULTICAST_LOOP option "
                  "on dgram for sending mcast messages.\n"));
      return TAO_ECG_Refcounted_Endpoint ();
    }

  if (this->non_blocking_
      && dgram.enable(ACE_NONBLOCK) == -1)
    {
      ACE_ERROR ((LM_ERROR,
                  "Error setting NON BLOCKING option.\n"));
      return TAO_ECG_Refcounted_Endpoint ();
    }

  return refendpoint;
}

TAO_ECG_Refcounted_Handler TAO_ECG_Mcast_Gateway::init_handler	(	TAO_ECG_Dgram_Handler *	recv,
		RtecEventChannelAdmin::EventChannel_ptr	ec,
		ACE_Reactor *	reactor
	)			[private]

Allocate and initialize appropriate objects.

Definition at line 417 of file ECG_Mcast_Gateway.cpp.

References ACE_ERROR, ACE_NEW_RETURN, address_server_arg_, ECG_HANDLER_BASIC, ECG_HANDLER_COMPLEX, ECG_HANDLER_UDP, ACE_String_Base< CHAR >::length(), LM_ERROR, nic_, TAO_ECG_UDP_EH::open(), TAO_ECG_Mcast_EH::open(), TAO_ECG_Simple_Mcast_EH::open(), ACE_Event_Handler::reactor(), ACE_Refcounted_Auto_Ptr< X, ACE_LOCK >::reset(), and ACE_INET_Addr::set().

{
  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.reset (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.reset (h);

      h->reactor (reactor);

      h->open (ec);
    }

  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.reset (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 ().\n"));
          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.\n"));
      return handler;
    }

  return handler;
}

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
	)			[private]

Allocate and initialize appropriate objects.

Definition at line 531 of file ECG_Mcast_Gateway.cpp.

References ACE_ES_EVENT_ANY, ACE_ES_EVENT_SOURCE_ANY, TAO_ECG_UDP_Receiver::create(), TAO_EC_Auto_Command< T >::disallow_command(), ACE_SupplierQOS_Factory::get_SupplierQOS(), TAO_EC_Servant_Var< T >::in(), ACE_SupplierQOS_Factory::insert(), RtecEventChannelAdmin::SupplierQOS::is_gateway, and TAO_EC_Auto_Command< T >::set_command().

Referenced by run().

{
  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);

  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 =
    const_cast<RtecEventChannelAdmin::SupplierQOS &> (supplier_qos_factory.get_SupplierQOS ());
  qos.is_gateway = 1;

  receiver->connect (qos);

  receiver_shutdown.disallow_command ();
  return receiver;
}

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
	)			[private]

Allocate and initialize appropriate objects.

Definition at line 488 of file ECG_Mcast_Gateway.cpp.

References ACE_ES_EVENT_ANY, ACE_ES_EVENT_SOURCE_ANY, TAO_ECG_UDP_Sender::create(), TAO_EC_Auto_Command< T >::disallow_command(), ACE_ConsumerQOS_Factory::get_ConsumerQOS(), TAO_EC_Servant_Var< T >::in(), ACE_ConsumerQOS_Factory::insert(), RtecEventChannelAdmin::ConsumerQOS::is_gateway, TAO_EC_Auto_Command< T >::set_command(), ACE_ConsumerQOS_Factory::start_disjunction_group(), and UDP_Sender_Shutdown.

Referenced by run().

{
  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);

  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_);
    }
  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 =
        const_cast<RtecEventChannelAdmin::ConsumerQOS &> (consumer_qos_factory.get_ConsumerQOS ());
      qos.is_gateway = 1;

      sender->connect (qos);
    }

  sender_shutdown.disallow_command ();
  return sender;
}

int TAO_ECG_Mcast_Gateway::init_svcs

(

void

)

[static]

Helper function to register the Gateway into the service configurator.

Definition at line 36 of file ECG_Mcast_Gateway.cpp.

References ACE_Service_Config::static_svcs().

{
  return ACE_Service_Config::static_svcs ()->
    insert (&ace_svc_desc_TAO_ECG_Mcast_Gateway);
}

void TAO_ECG_Mcast_Gateway::run	(	CORBA::ORB_ptr	orb,
		RtecEventChannelAdmin::EventChannel_ptr	ec
	)

The main method - create, configure and run federation components according to the specified configuration.

Definition at line 583 of file ECG_Mcast_Gateway.cpp.

References ACE_DEBUG, activate(), TAO_EC_Auto_Command< T >::disallow_command(), TAO_EC_Object_Deactivator::disallow_deactivation(), ECG_MCAST_RECEIVER, ECG_MCAST_SENDER, ECG_MCAST_TWO_WAY, ACE_Refcounted_Auto_Ptr< X, ACE_LOCK >::get(), TAO_EC_Servant_Var< T >::in(), TAO_Objref_Var_T< T >::in(), PortableServer::Servant_var< T >::in(), init_address_server(), init_endpoint(), init_receiver(), init_sender(), LM_ERROR, CORBA::ORB::orb_core(), TAO_ORB_Core::reactor(), ACE_Event_Handler::reactor(), TAO_EC_Auto_Command< T >::set_command(), UDP_Sender_Shutdown, and verify_args().

{
  // Verify args.
  this->verify_args (orb, ec);

  // 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.\n"));
      throw CORBA::INTERNAL ();
    }

  RtecUDPAdmin::AddrServer_var address_server;

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

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

  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)
        {
          throw CORBA::INTERNAL ();
        }

      sender = this->init_sender (ec,
                                  address_server.in (),
                                  endpoint_rptr);
      if (!sender.in ())
        {
          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);
      if (!receiver.in ())
        {
          throw CORBA::INTERNAL ();
        }

      receiver_shutdown.set_command (UDP_Receiver_Shutdown (receiver));

      TAO_ECG_Refcounted_Handler
        handler_rptr (this->init_handler (receiver.in (),
                                          ec,
                                          reactor));
      if (handler_rptr.get () == 0)
        {
          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 ();
}

int TAO_ECG_Mcast_Gateway::validate_configuration

(

void

)

[private]

Verifies configuration values specified through init() make sense.

Definition at line 244 of file ECG_Mcast_Gateway.cpp.

References ACE_DEBUG, ECG_ADDRESS_SERVER_BASIC, ECG_HANDLER_BASIC, ECG_HANDLER_UDP, ECG_MCAST_SENDER, and LM_ERROR.

Referenced by init().

{
  if ((this->handler_type_ == ECG_HANDLER_BASIC
       || this->handler_type_ == ECG_HANDLER_UDP)
      && this->service_type_ != ECG_MCAST_SENDER
      && this->address_server_type_ != ECG_ADDRESS_SERVER_BASIC)
    {
      ACE_DEBUG ((LM_ERROR,
                      "Configurations for mcast handler and "
                      "address server do not match.\n"));
      return -1;
    }

  // Currently all Address Server implementations require an
  // initialization string.  If we ever add a new Address Server
  // implementation, which does not, we'll have to remove this check.
  if (this->address_server_arg_.length () == 0)
    {
      ACE_DEBUG ((LM_ERROR,
                      "Address server initializaton "
                      "argument not specified.\n"));
      return -1;
    }

  if (this->ip_multicast_loop_ != 0
      && this->ip_multicast_loop_ != 1)
    {
      ACE_DEBUG ((LM_ERROR,
                  "IP MULTICAST LOOP option must have a boolean value.\n"));
      return -1;
    }

  if (this->non_blocking_ != 0
      && this->non_blocking_ != 1)
    {
      ACE_DEBUG ((LM_ERROR,
                  "NON BLOCKING flag must have a boolean value.\n"));
      return -1;
    }

  return 0;
}

void TAO_ECG_Mcast_Gateway::verify_args	(	CORBA::ORB_ptr	orb,
		RtecEventChannelAdmin::EventChannel_ptr	ec
	)			[private]

Check that arguments to run() are not nil.

Definition at line 563 of file ECG_Mcast_Gateway.cpp.

References ACE_ERROR, CORBA::is_nil(), and LM_ERROR.

Referenced by run().

{
  if (CORBA::is_nil (ec))
    {
      ACE_ERROR ((LM_ERROR,
                  "Nil event channel argument passed to "
                  "TAO_ECG_Mcast_Gateway::run().\n"));
      throw CORBA::INTERNAL ();
    }
  if (CORBA::is_nil (orb))
    {
      ACE_ERROR ((LM_ERROR,
                  "Nil orb argument passed to "
                  "TAO_ECG_Mcast_Gateway::run().\n"));
      throw CORBA::INTERNAL ();
    }
}

---

Member Data Documentation

ACE_CString TAO_ECG_Mcast_Gateway::address_server_arg_ [private]

Flags controlling configuration.

Definition at line 265 of file ECG_Mcast_Gateway.h.

Referenced by init(), init_address_server(), and init_handler().

Address_Server_Type TAO_ECG_Mcast_Gateway::address_server_type_ [private]

Flags controlling configuration.

Definition at line 264 of file ECG_Mcast_Gateway.h.

Referenced by init().

RtecEventChannelAdmin::ConsumerQOS TAO_ECG_Mcast_Gateway::consumer_qos_ [private]

Flags controlling configuration.

Definition at line 271 of file ECG_Mcast_Gateway.h.

Referenced by init().

Handler_Type TAO_ECG_Mcast_Gateway::handler_type_ [private]

Flags controlling configuration.

Definition at line 263 of file ECG_Mcast_Gateway.h.

Referenced by init().

int TAO_ECG_Mcast_Gateway::ip_multicast_loop_ [private]

Flags controlling configuration.

Definition at line 268 of file ECG_Mcast_Gateway.h.

Referenced by init().

ACE_CString TAO_ECG_Mcast_Gateway::nic_ [private]

Flags controlling configuration.

Definition at line 267 of file ECG_Mcast_Gateway.h.

Referenced by init(), and init_handler().

int TAO_ECG_Mcast_Gateway::non_blocking_ [private]

Flags controlling configuration.

Definition at line 269 of file ECG_Mcast_Gateway.h.

Referenced by init().

Service_Type TAO_ECG_Mcast_Gateway::service_type_ [private]

Flags controlling configuration.

Definition at line 262 of file ECG_Mcast_Gateway.h.

Referenced by init().

u_char TAO_ECG_Mcast_Gateway::ttl_value_ [private]

Flags controlling configuration.

Definition at line 266 of file ECG_Mcast_Gateway.h.

Referenced by init().

---

The documentation for this class was generated from the following files:

• ECG_Mcast_Gateway.h
• ECG_Mcast_Gateway.cpp

---