TAO/orbsvcs/examples/RtEC/IIOPGateway/Consumer.cpp

   1 #include "Consumer.h"
   2 #include "orbsvcs/RtecEventChannelAdminC.h"
   3 #include "orbsvcs/Event_Service_Constants.h"
   4 #include "orbsvcs/Event_Utilities.h"
   5 #include "orbsvcs/CosNamingC.h"
   6 #include "ace/Arg_Shifter.h"
   7 #include "ace/OS_NS_string.h"
   8
   9 const RtecEventComm::EventSourceID MY_SOURCE_ID  = ACE_ES_EVENT_SOURCE_ANY + 1;
  10 const RtecEventComm::EventType     MY_EVENT_TYPE = ACE_ES_EVENT_UNDEFINED + 1;
  11
  12 static const ACE_TCHAR *ecname = 0;
  13
  14 int
  15 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
  16 {
  17   Consumer consumer;
  18
  19   return consumer.run (argc, argv);
  20 }
  21
  22 // ****************************************************************
  23
  24 Consumer::Consumer ()
  25   : event_count_ (0)
  26 {
  27 }
  28
  29 int
  30 Consumer::run (int argc, ACE_TCHAR* argv[])
  31 {
  32   try
  33     {
  34       // First parse our command line options
  35       if (this->parse_args(argc, argv) != 0)
  36       {
  37          return -1;
  38       }
  39
  40       // ORB initialization boiler plate...
  41       CORBA::ORB_var orb =
  42         CORBA::ORB_init (argc, argv);
  43
  44       // Do *NOT* make a copy because we don't want the ORB to outlive
  45       // the run() method.
  46       this->orb_ = orb.in ();
  47
  48       CORBA::Object_var object =
  49         orb->resolve_initial_references ("RootPOA");
  50       PortableServer::POA_var poa =
  51         PortableServer::POA::_narrow (object.in ());
  52       PortableServer::POAManager_var poa_manager =
  53         poa->the_POAManager ();
  54       poa_manager->activate ();
  55
  56       // Obtain the event channel from the naming service
  57       CORBA::Object_var naming_obj =
  58         orb->resolve_initial_references ("NameService");
  59
  60       if (CORBA::is_nil (naming_obj.in ()))
  61         ACE_ERROR_RETURN ((LM_ERROR,
  62                            " (%P|%t) Unable to get the Naming Service.\n"),
  63                           1);
  64
  65       CosNaming::NamingContext_var naming_context =
  66         CosNaming::NamingContext::_narrow (naming_obj.in ());
  67
  68       CosNaming::Name name (1);
  69       name.length (1);
  70       name[0].id = CORBA::string_dup (ACE_TEXT_ALWAYS_CHAR(ecname));
  71
  72       CORBA::Object_var ec_obj =
  73         naming_context->resolve (name);
  74
  75       RtecEventChannelAdmin::EventChannel_var event_channel =
  76         RtecEventChannelAdmin::EventChannel::_narrow (ec_obj.in ());
  77
  78       if (CORBA::is_nil (event_channel.in ()))
  79         ACE_ERROR_RETURN ((LM_ERROR,
  80                            " (%P|%t) Unable to get Event Channel.\n"),
  81                           1);
  82
  83       // The canonical protocol to connect to the EC
  84       RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
  85         event_channel->for_consumers ();
  86
  87       RtecEventChannelAdmin::ProxyPushSupplier_var supplier =
  88         consumer_admin->obtain_push_supplier ();
  89
  90       RtecEventComm::PushConsumer_var consumer =
  91         this->_this ();
  92
  93       ACE_ConsumerQOS_Factory qos;
  94       qos.start_disjunction_group ();
  95       qos.insert (MY_SOURCE_ID,   // Source ID
  96                   MY_EVENT_TYPE,  // Event Type
  97                   0);             // handle to the rt_info
  98       for (int i = 0; i < 10; i++)
  99         {
 100           qos.insert (MY_SOURCE_ID  + i,   // Source ID
 101                       MY_EVENT_TYPE + i,  // Event Type
 102                       0);             // handle to the rt_info
 103         }
 104       supplier->connect_push_consumer (consumer.in (), qos);
 105
 106       // Wait for events, using work_pending()/perform_work() may help
 107       // or using another thread, this example is too simple for that.
 108       orb->run ();
 109
 110       // We don't do any cleanup, it is hard to do it after shutdown,
 111       // and would complicate the example; plus it is almost
 112       // impossible to do cleanup after ORB->run() because the POA is
 113       // in the holding state.  Applications should use
 114       // work_pending()/perform_work() to do more interesting stuff.
 115       // Check the supplier for the proper way to do cleanup.
 116     }
 117   catch (const CORBA::Exception& ex)
 118     {
 119       ex._tao_print_exception ("Consumer::run");
 120       return 1;
 121     }
 122   return 0;
 123 }
 124
 125 void
 126 Consumer::push (const RtecEventComm::EventSet& events)
 127 {
 128   if (events.length () == 0)
 129     {
 130       ACE_DEBUG ((LM_DEBUG,
 131                   "Consumer (%P|%t) no events\n"));
 132       return;
 133     }
 134
 135   this->event_count_ += events.length ();
 136   if (this->event_count_ % 100 == 0)
 137     {
 138       ACE_DEBUG ((LM_DEBUG,
 139                   "Consumer (%P|%t): %d events received\n",
 140                   this->event_count_));
 141     }
 142 }
 143
 144 void
 145 Consumer::disconnect_push_consumer ()
 146 {
 147   // In this example we shutdown the ORB when we disconnect from the
 148   // EC (or rather the EC disconnects from us), but this doesn't have
 149   // to be the case....
 150   this->orb_->shutdown (false);
 151 }
 152
 153 int
 154 Consumer::parse_args (int argc, ACE_TCHAR *argv[])
 155 {
 156   ACE_Arg_Shifter arg_shifter (argc, argv);
 157
 158   while (arg_shifter.is_anything_left ())
 159     {
 160       const ACE_TCHAR *arg = arg_shifter.get_current ();
 161
 162       if (ACE_OS::strcmp (arg, ACE_TEXT("-e")) == 0)
 163         {
 164           arg_shifter.consume_arg ();
 165           ecname = arg_shifter.get_current ();
 166         }
 167
 168       arg_shifter.ignore_arg ();
 169     }
 170
 171   // Indicates successful parsing of the command line
 172   return 0;
 173 }
 174