TAO/orbsvcs/tests/Event/lib/Consumer.cpp

   1 #include "Consumer.h"
   2 #include "orbsvcs/Event_Service_Constants.h"
   3 #include "orbsvcs/Time_Utilities.h"
   4
   5 #include "tao/debug.h"
   6
   7 EC_Consumer::EC_Consumer (EC_Driver *driver,
   8                           void *cookie)
   9   : driver_ (driver),
  10     cookie_ (cookie),
  11     push_count_ (0),
  12     shutdown_event_type_ (ACE_ES_EVENT_SHUTDOWN),
  13     is_active_ (0)
  14 {
  15 }
  16
  17 void
  18 EC_Consumer::connect (
  19     RtecEventChannelAdmin::ConsumerAdmin_ptr consumer_admin,
  20     const RtecEventChannelAdmin::ConsumerQOS& qos,
  21     int shutdown_event_type)
  22 {
  23   this->supplier_proxy_ =
  24     consumer_admin->obtain_push_supplier ();
  25
  26   this->connect (qos, shutdown_event_type);
  27 }
  28
  29 void
  30 EC_Consumer::connect (
  31     const RtecEventChannelAdmin::ConsumerQOS& qos,
  32     int shutdown_event_type)
  33 {
  34   if (CORBA::is_nil (this->supplier_proxy_.in ()))
  35     return; // @@ Throw?
  36
  37   this->shutdown_event_type_ = shutdown_event_type;
  38
  39   if (CORBA::is_nil (this->myself_.in ()))
  40     {
  41       this->myself_ = this->_this ();
  42     }
  43   this->is_active_ = 1;
  44
  45   this->supplier_proxy_->connect_push_consumer (this->myself_.in (),
  46                                                 qos);
  47 }
  48
  49 int
  50 EC_Consumer::connected () const
  51 {
  52   return !CORBA::is_nil (this->supplier_proxy_.in ());
  53 }
  54
  55 void
  56 EC_Consumer::disconnect ()
  57 {
  58   if (CORBA::is_nil (this->supplier_proxy_.in ()))
  59     return;
  60
  61   this->supplier_proxy_->disconnect_push_supplier ();
  62
  63   this->supplier_proxy_ =
  64     RtecEventChannelAdmin::ProxyPushSupplier::_nil ();
  65 }
  66
  67 void
  68 EC_Consumer::shutdown ()
  69 {
  70   if (!this->is_active_)
  71     return;
  72
  73   // Deactivate the servant
  74   PortableServer::POA_var poa =
  75     this->_default_POA ();
  76   PortableServer::ObjectId_var id =
  77     poa->servant_to_id (this);
  78   poa->deactivate_object (id.in ());
  79   this->myself_ = RtecEventComm::PushConsumer::_nil ();
  80   this->is_active_ = 0;
  81 }
  82
  83 void
  84 EC_Consumer::dump_results (const ACE_TCHAR* name,
  85                            ACE_High_Res_Timer::global_scale_factor_type gsf)
  86 {
  87   this->throughput_.dump_results (name, gsf);
  88 }
  89
  90 void
  91 EC_Consumer::accumulate (ACE_Throughput_Stats& throughput) const
  92 {
  93   throughput.accumulate (this->throughput_);
  94 }
  95
  96 void
  97 EC_Consumer::push (const RtecEventComm::EventSet& events)
  98 {
  99   this->driver_->consumer_push (this->cookie_, events);
 100
 101   if (events.length () == 0)
 102     {
 103       ACE_DEBUG ((LM_DEBUG,
 104                   "EC_Consumer (%P|%t) no events\n"));
 105       return;
 106     }
 107
 108   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 109   if (this->push_count_ == 0)
 110     this->throughput_start_ = ACE_OS::gethrtime ();
 111
 112   this->push_count_ += events.length ();
 113
 114   if (TAO_debug_level > 0
 115       && this->push_count_ % 100 == 0)
 116     {
 117       ACE_DEBUG ((LM_DEBUG,
 118                   "EC_Consumer (%P|%t): %d events received\n",
 119                   this->push_count_));
 120     }
 121
 122   for (u_int i = 0; i < events.length (); ++i)
 123     {
 124       const RtecEventComm::Event& e = events[i];
 125
 126       ACE_hrtime_t creation;
 127       ORBSVCS_Time::TimeT_to_hrtime (creation,
 128                                      e.header.creation_time);
 129
 130       const ACE_hrtime_t now = ACE_OS::gethrtime ();
 131       this->throughput_.sample (now - this->throughput_start_,
 132                                 now - creation);
 133
 134       if (e.header.type == this->shutdown_event_type_)
 135         this->driver_->consumer_shutdown (this->cookie_);
 136     }
 137 }
 138
 139 void
 140 EC_Consumer::disconnect_push_consumer ()
 141 {
 142   this->driver_->consumer_disconnect (this->cookie_);
 143   this->supplier_proxy_ =
 144     RtecEventChannelAdmin::ProxyPushSupplier::_nil ();
 145 }