TAO/orbsvcs/tests/CosEvent/lib/Counting_Consumer.cpp

   1 #include "Counting_Consumer.h"
   2 #include "ace/OS_NS_unistd.h"
   3
   4 CEC_Counting_Consumer::CEC_Counting_Consumer (const char* name)
   5   : event_count (0),
   6     disconnect_count (0),
   7     name_ (name)
   8 {
   9 }
  10
  11 void
  12 CEC_Counting_Consumer::connect (CosEventChannelAdmin::ConsumerAdmin_ptr consumer_admin)
  13 {
  14   // The canonical protocol to connect to the EC
  15
  16   CosEventComm::PushConsumer_var consumer =
  17     this->_this ();
  18
  19   if (CORBA::is_nil (this->supplier_proxy_.in ()))
  20     {
  21       this->supplier_proxy_ =
  22         consumer_admin->obtain_push_supplier ();
  23     }
  24
  25   this->supplier_proxy_->connect_push_consumer (consumer.in ());
  26 }
  27
  28 void
  29 CEC_Counting_Consumer::disconnect ()
  30 {
  31   if (!CORBA::is_nil (this->supplier_proxy_.in ()))
  32     {
  33       this->supplier_proxy_->disconnect_push_supplier ();
  34     }
  35
  36   PortableServer::POA_var consumer_poa =
  37     this->_default_POA ();
  38   PortableServer::ObjectId_var consumer_id =
  39     consumer_poa->servant_to_id (this);
  40   consumer_poa->deactivate_object (consumer_id.in ());
  41
  42   this->supplier_proxy_ =
  43     CosEventChannelAdmin::ProxyPushSupplier::_nil ();
  44 }
  45
  46 void
  47 CEC_Counting_Consumer::dump_results (int expected_count, int tolerance)
  48 {
  49   int diff = this->event_count - expected_count;
  50   if (diff > tolerance || diff < -tolerance)
  51     {
  52       ACE_DEBUG ((LM_DEBUG,
  53                   "ERROR - %s unexpected number of events  <%d>\n",
  54                   this->name_,
  55                   this->event_count));
  56     }
  57   else
  58     {
  59       ACE_DEBUG ((LM_DEBUG,
  60                   "%s - number of events <%d> within margins\n",
  61                   this->name_,
  62                   this->event_count));
  63     }
  64 }
  65
  66 void
  67 CEC_Counting_Consumer::push (const CORBA::Any&)
  68 {
  69   this->event_count ++;
  70 #if 0
  71   if (this->event_count % 10 == 0)
  72     {
  73       ACE_DEBUG ((LM_DEBUG,
  74                   "%s (%P|%t): %d events received\n",
  75                   this->name_,
  76                   this->event_count));
  77     }
  78 #endif /* 0 */
  79 }
  80
  81 void
  82 CEC_Counting_Consumer::disconnect_push_consumer ()
  83 {
  84   this->disconnect_count++;
  85   this->supplier_proxy_ =
  86     CosEventChannelAdmin::ProxyPushSupplier::_nil ();
  87 }
  88
  89 // ****************************************************************
  90
  91 CEC_Pull_Counting_Consumer::CEC_Pull_Counting_Consumer (const char* name)
  92   : event_count (0),
  93     disconnect_count (0),
  94     name_ (name)
  95 {
  96 }
  97
  98 void
  99 CEC_Pull_Counting_Consumer::connect (CosEventChannelAdmin::ConsumerAdmin_ptr consumer_admin)
 100 {
 101   // The canonical protocol to connect to the EC
 102
 103   CosEventComm::PullConsumer_var consumer =
 104     this->_this ();
 105
 106   if (CORBA::is_nil (this->supplier_proxy_.in ()))
 107     {
 108       this->supplier_proxy_ =
 109         consumer_admin->obtain_pull_supplier ();
 110     }
 111
 112   this->supplier_proxy_->connect_pull_consumer (consumer.in ());
 113 }
 114
 115 void
 116 CEC_Pull_Counting_Consumer::disconnect ()
 117 {
 118   if (!CORBA::is_nil (this->supplier_proxy_.in ()))
 119     {
 120       this->supplier_proxy_->disconnect_pull_supplier ();
 121     }
 122
 123   PortableServer::POA_var consumer_poa =
 124     this->_default_POA ();
 125   PortableServer::ObjectId_var consumer_id =
 126     consumer_poa->servant_to_id (this);
 127   consumer_poa->deactivate_object (consumer_id.in ());
 128
 129   this->supplier_proxy_ =
 130     CosEventChannelAdmin::ProxyPullSupplier::_nil ();
 131 }
 132
 133 CORBA::Any*
 134 CEC_Pull_Counting_Consumer::pull ()
 135 {
 136   if (CORBA::is_nil (this->supplier_proxy_.in ()))
 137     {
 138       return 0;
 139     }
 140   this->event_count++;
 141   return this->supplier_proxy_->pull ();
 142 }
 143
 144 CORBA::Any*
 145 CEC_Pull_Counting_Consumer::try_pull (CORBA::Boolean_out has_event)
 146 {
 147   if (CORBA::is_nil (this->supplier_proxy_.in ()))
 148     {
 149       has_event = 0;
 150       return 0;
 151     }
 152
 153   CORBA::Any_var event =
 154     this->supplier_proxy_->try_pull (has_event);
 155
 156   if (has_event)
 157     this->event_count++;
 158
 159   return event._retn ();
 160 }
 161
 162 void
 163 CEC_Pull_Counting_Consumer::dump_results (int expected_count, int tolerance)
 164 {
 165   int diff = this->event_count - expected_count;
 166   if (diff > tolerance || diff < -tolerance)
 167     {
 168       ACE_DEBUG ((LM_DEBUG,
 169                   "ERROR - %s unexpected number of events  <%d>\n",
 170                   this->name_,
 171                   this->event_count));
 172     }
 173   else
 174     {
 175       ACE_DEBUG ((LM_DEBUG,
 176                   "%s - number of events <%d> within margins\n",
 177                   this->name_,
 178                   this->event_count));
 179     }
 180 }
 181
 182 void
 183 CEC_Pull_Counting_Consumer::disconnect_pull_consumer ()
 184 {
 185   this->disconnect_count++;
 186   this->supplier_proxy_ =
 187     CosEventChannelAdmin::ProxyPullSupplier::_nil ();
 188 }
 189
 190 // ****************************************************************
 191
 192 CEC_Counting_Consumer_Task::
 193     CEC_Counting_Consumer_Task (CEC_Pull_Counting_Consumer *s,
 194                                 int milliseconds)
 195   :  consumer_ (s),
 196      stop_flag_ (0),
 197      pull_count_ (0),
 198      milliseconds_ (milliseconds)
 199 {
 200 }
 201
 202 int
 203 CEC_Counting_Consumer_Task::svc ()
 204 {
 205   try
 206     {
 207       this->run ();
 208     }
 209   catch (const CORBA::Exception&)
 210     {
 211       return -1;
 212     }
 213   return 0;
 214 }
 215
 216 void
 217 CEC_Counting_Consumer_Task::stop ()
 218 {
 219   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 220   this->stop_flag_ = 1;
 221 }
 222
 223 CORBA::ULong
 224 CEC_Counting_Consumer_Task::pull_count ()
 225 {
 226   return this->pull_count_;
 227 }
 228
 229 void
 230 CEC_Counting_Consumer_Task::run ()
 231 {
 232   CORBA::Any event;
 233   event <<= CORBA::Long(0);
 234
 235   int stop = 0;
 236   do {
 237     CORBA::Boolean has_event;
 238     CORBA::Any_var event =
 239       this->consumer_->try_pull (has_event);
 240
 241     if (this->milliseconds_ != 0)
 242       {
 243         ACE_Time_Value tv (0, 1000 * this->milliseconds_);
 244         ACE_OS::sleep (tv);
 245       }
 246     else
 247       {
 248         // Sleep for a short time to avoid spinning... kind of klugy
 249         ACE_OS::sleep (0);
 250       }
 251
 252     if (has_event)
 253       {
 254         ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 255         this->pull_count_++;
 256       }
 257
 258     stop = this->stop_flag_;
 259   } while (stop == 0);
 260 }