diff options
Diffstat (limited to 'trunk/TAO/orbsvcs/tests/Event/UDP/receiver.cpp')
| -rw-r--r-- | trunk/TAO/orbsvcs/tests/Event/UDP/receiver.cpp | 366 |
1 files changed, 366 insertions, 0 deletions
diff --git a/trunk/TAO/orbsvcs/tests/Event/UDP/receiver.cpp b/trunk/TAO/orbsvcs/tests/Event/UDP/receiver.cpp new file mode 100644 index 00000000000..30a675ede38 --- /dev/null +++ b/trunk/TAO/orbsvcs/tests/Event/UDP/receiver.cpp @@ -0,0 +1,366 @@ +// $Id$ + + +#include "Consumer.h" +#include "AddrServer.h" +#include "orbsvcs/Event_Service_Constants.h" +#include "orbsvcs/Event/EC_Event_Channel.h" +#include "orbsvcs/Event/EC_Default_Factory.h" +#include "orbsvcs/Event/ECG_Mcast_EH.h" +#include "orbsvcs/Event/ECG_UDP_Receiver.h" +#include "orbsvcs/Event/ECG_UDP_Out_Endpoint.h" +#include "tao/ORB_Core.h" +#include "ace/Get_Opt.h" +#include "ace/OS_NS_unistd.h" +#include "TestC.h" + +ACE_RCSID (EC_Examples, + MCast, + "$Id$") + +const char *udp_mcast_address = + ACE_DEFAULT_MULTICAST_ADDR ":10001"; +bool valuetype = false; + +int parse_args (int argc, char *argv[]); + +int +main (int argc, char* argv[]) +{ + // Register the default factory in the Service Configurator. + // If your platform supports static constructors then you can + // simply using the ACE_STATIC_SVC_DEFINE() macro, unfortunately TAO + // must run on platforms where static constructors do not work well, + // so we have to explicitly invoke this function. + TAO_EC_Default_Factory::init_svcs (); + + // The exception macros are described in $ACE_ROOT/docs/exceptions.html + // and defined in $ACE_ROOT/ace/CORBA_macros.h. + // If your platform supports native exceptions, and TAO was compiled + // with native exception support then you can simply use try/catch + // and avoid the ACE_ENV_SINGLE_ARG_PARAMETER argument. + // Unfortunately many embedded systems cannot use exceptions due to + // the space and time overhead. + // + ACE_DECLARE_NEW_CORBA_ENV; + ACE_TRY + { + // **************** HERE STARTS THE ORB SETUP + + // Create the ORB, pass the argv list for parsing. + CORBA::ORB_var orb = + CORBA::ORB_init (argc, argv, "" ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + + // Parse the arguments, you usually want to do this after + // invoking ORB_init() because ORB_init() will remove all the + // -ORB options from the command line. + if (parse_args (argc, argv) == -1) + { + ACE_ERROR ((LM_ERROR, + "Usage: Service [-m udp_mcast_addr]\n")); + return 1; + } + + if (valuetype) + { + Hello::ValueTypeData_init *vb_factory = 0; + ACE_NEW_RETURN (vb_factory, + Hello::ValueTypeData_init, + 1); // supplied by mapping + + orb->register_value_factory (vb_factory->tao_repository_id (), + vb_factory + ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + vb_factory->_remove_ref (); // release ownership + } + + // This is the standard code to get access to the POA and + // activate it. + // The POA starts in the holding state, if it is not activated + // it will not process any requests. + CORBA::Object_var object = + orb->resolve_initial_references ("RootPOA" ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + PortableServer::POA_var poa = + PortableServer::POA::_narrow (object.in () ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + PortableServer::POAManager_var poa_manager = + poa->the_POAManager (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + poa_manager->activate (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + + // **************** THAT COMPLETS THE ORB SETUP + + // **************** HERE START THE LOCAL EVENT CHANNEL SETUP + + // This structure is used to define the startup time event + // channel configuration. + // This structure is described in + // + // $TAO_ROOT/docs/ec_options.html + // + TAO_EC_Event_Channel_Attributes attributes (poa.in (), + poa.in ()); + + // Create the Event Channel implementation class + TAO_EC_Event_Channel ec_impl (attributes); + + // Activate the Event Channel, depending on the configuration + // that may involve creating some threads. + // But it should always be invoked because several internal data + // structures are initialized at that point. + ec_impl.activate (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + + // The event channel is activated as any other CORBA servant. + // In this case we use the simple implicit activation with the + // RootPOA + RtecEventChannelAdmin::EventChannel_var event_channel = + ec_impl._this (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + + // **************** THAT COMPLETES THE LOCAL EVENT CHANNEL SETUP + + // **************** HERE STARTS THE FEDERATION SETUP + + // The next step is to setup the multicast gateways. + // There are two gateways involved, one sends the locally + // generated events to the federated peers, the second gateway + // receives multicast traffic and turns it into local events. + + // The sender requires a helper object to select what + // multicast group will carry what traffic, this is the + // so-called 'Address Server'. + // The intention is that advanced applications can use different + // multicast groups for different events, this can exploit + // network interfaces that filter unwanted multicast traffic. + // The helper object is accessed through an IDL interface, so it + // can reside remotely. + // In this example, and in many application, using a fixed + // multicast group is enough, and a local address server is the + // right approach. + + // First we convert the string into an INET address, then we + // convert that into the right IDL structure: + ACE_INET_Addr udp_addr (udp_mcast_address); + ACE_DEBUG ((LM_DEBUG, + "udp mcast address is: %s\n", + udp_mcast_address)); + RtecUDPAdmin::UDP_Addr addr; + addr.ipaddr = udp_addr.get_ip_address (); + addr.port = udp_addr.get_port_number (); + + // Now we create and activate the servant + AddrServer as_impl (addr); + RtecUDPAdmin::AddrServer_var address_server = + as_impl._this (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + + TAO_ECG_Refcounted_Endpoint endpoint(new TAO_ECG_UDP_Out_Endpoint); + + // Now we connect the sender as a consumer of events, it will + // receive any event from any source and send it to the "right" + // multicast group, as defined by the address server set above: + RtecEventChannelAdmin::ConsumerQOS sub; + sub.is_gateway = 1; + + sub.dependencies.length (1); + sub.dependencies[0].event.header.type = + ACE_ES_EVENT_ANY; // first free event type + sub.dependencies[0].event.header.source = + ACE_ES_EVENT_SOURCE_ANY; // Any source is OK + + // To receive events we need to setup an event handler: + TAO_EC_Servant_Var<TAO_ECG_UDP_Receiver> receiver = TAO_ECG_UDP_Receiver::create(); + TAO_ECG_Mcast_EH mcast_eh (&(*receiver)); + + // The event handler uses the ORB reactor to wait for multicast + // traffic: + mcast_eh.reactor (orb->orb_core ()->reactor ()); + + // The multicast Event Handler needs to know to what multicast + // groups it should listen to. To do so it becomes an observer + // with the event channel, to determine the list of events + // required by all the local consumer. + // Then it register for the multicast groups that carry those + // events: + mcast_eh.open (event_channel.in () + ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + + // Again the receiver connects to the event channel as a + // supplier of events, using the Observer features to detect + // local consumers and their interests: + receiver->init (event_channel.in (), + endpoint, + address_server.in () + ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + + // The Receiver is also a supplier of events. The exact type of + // events is only known to the application, because it depends + // on the traffic carried by all the multicast groups that the + // different event handlers subscribe to. + // In this example we choose to simply describe our publications + // using wilcards, any event from any source. More advanced + // application could use the Observer features in the event + // channel to update this information (and reduce the number of + // multicast groups that each receive subscribes to). + // In a future version the event channel could perform some of + // those tasks automatically + RtecEventChannelAdmin::SupplierQOS pub; + pub.publications.length (1); + pub.publications[0].event.header.type = ACE_ES_EVENT_ANY; + pub.publications[0].event.header.source = ACE_ES_EVENT_SOURCE_ANY; + pub.is_gateway = 1; + + receiver->connect (pub ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + + // **************** THAT COMPLETES THE FEDERATION SETUP + + // **************** HERE STARTS THE CLIENT SETUP + + // First let us create a consumer and connect it to the event + // channel + Consumer consumer (valuetype); + RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin = + event_channel->for_consumers (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + consumer.connect (consumer_admin.in () + ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + + // **************** THAT COMPLETES THE CLIENT SETUP + + // **************** HERE STARTS THE EVENT LOOP + + // Wait for events, including incoming multicast data. + // We could also use orb->run(), but that will not let us + // terminate the application in a nice way. + for (int i = 0; i != 100; ++i) + { + CORBA::Boolean there_is_work = + orb->work_pending (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + if (there_is_work) + { + // We use a TAO extension. The CORBA mechanism does not + // provide any decent way to control the duration of + // perform_work() or work_pending(), so just calling + // them results in a spin loop. + ACE_Time_Value tv (0, 50000); + orb->perform_work (tv ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + } + ACE_Time_Value tv (0, 100000); + ACE_OS::sleep (tv); + if (consumer.event_count () == 25) + { + break; + } + } + + // **************** THAT COMPLETES THE EVENT LOOP + + // **************** HERE STARTS THE CLEANUP CODE + + consumer.disconnect (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + + // Now let us close the Receiver + receiver->shutdown (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + + int const r = mcast_eh.shutdown (); + + if (r == -1) + { + ACE_ERROR_RETURN ((LM_ERROR, + "Closing MCast event handler\n"), 1); + } + + // The event channel must be destroyed, so it can release its + // resources, and inform all the clients that are still + // connected that it is going away. + event_channel->destroy (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + + // Deactivating the event channel implementation is not strictly + // required, the POA will do it for us, but it is good manners: + { + // Using _this() activates with the default POA, we must gain + // access to that POA to deactivate the object. + // Notice that we 'know' that the default POA for this servant + // is the root POA, but the code is more robust if we don't + // rely on that. + PortableServer::POA_var poa = + ec_impl._default_POA (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + // Get the Object Id used for the servant.. + PortableServer::ObjectId_var oid = + poa->servant_to_id (&ec_impl ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + // Deactivate the object + poa->deactivate_object (oid.in () ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + } + + // Now we can destroy the POA, the flags mean that we want to + // wait until the POA is really destroyed + poa->destroy (1, 1 ACE_ENV_ARG_PARAMETER); + ACE_TRY_CHECK; + + // Finally destroy the ORB + orb->destroy (ACE_ENV_SINGLE_ARG_PARAMETER); + ACE_TRY_CHECK; + + // **************** THAT COMPLETES THE CLEANUP CODE + + ACE_DEBUG ((LM_DEBUG, + "UDP receiver ready\n")); + } + ACE_CATCHANY + { + ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "Service"); + return 1; + } + ACE_ENDTRY; + return 0; +} + +// **************************************************************** + +int parse_args (int argc, char *argv[]) +{ + ACE_Get_Opt get_opts (argc, argv, "vm:"); + int c; + + while ((c = get_opts ()) != -1) + switch (c) + { + case 'm': + udp_mcast_address = get_opts.opt_arg (); + break; + + case 'v': + valuetype = true; + break; + + case '?': + default: + ACE_ERROR_RETURN ((LM_ERROR, + "usage: %s " + "[-m udp_mcast_address]" + "[-v]" + "\n", + argv [0]), + -1); + } + // Indicates sucessful parsing of the command line + return 0; +} + |