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

   1 #include "Counting_Supplier.h"
   2 #include "ace/OS_NS_unistd.h"
   3
   4
   5 CEC_Counting_Supplier::CEC_Counting_Supplier ()
   6   :  event_count (0),
   7      disconnect_count (0)
   8 {
   9 }
  10
  11 void
  12 CEC_Counting_Supplier::connect (
  13     CosEventChannelAdmin::SupplierAdmin_ptr supplier_admin)
  14 {
  15   CosEventComm::PushSupplier_var supplier =
  16     this->_this ();
  17
  18   if (CORBA::is_nil (this->consumer_proxy_.in ()))
  19     {
  20       this->consumer_proxy_ =
  21         supplier_admin->obtain_push_consumer ();
  22     }
  23
  24   this->consumer_proxy_->connect_push_supplier (supplier.in ());
  25 }
  26
  27 void
  28 CEC_Counting_Supplier::disconnect ()
  29 {
  30   if (!CORBA::is_nil (this->consumer_proxy_.in ()))
  31     {
  32       this->consumer_proxy_->disconnect_push_consumer ();
  33     }
  34
  35   PortableServer::POA_var supplier_poa =
  36     this->_default_POA ();
  37   PortableServer::ObjectId_var supplier_id =
  38     supplier_poa->servant_to_id (this);
  39   supplier_poa->deactivate_object (supplier_id.in ());
  40
  41   this->consumer_proxy_ =
  42     CosEventChannelAdmin::ProxyPushConsumer::_nil ();
  43 }
  44
  45 void
  46 CEC_Counting_Supplier::push (const CORBA::Any&)
  47 {
  48   if (CORBA::is_nil (this->consumer_proxy_.in ()))
  49     return;
  50
  51   CORBA::Any event;
  52   event <<= CORBA::Long(0);
  53
  54   this->consumer_proxy_->push (event);
  55   this->event_count++;
  56 }
  57
  58 void
  59 CEC_Counting_Supplier::disconnect_push_supplier ()
  60 {
  61   this->disconnect_count++;
  62   this->consumer_proxy_ =
  63     CosEventChannelAdmin::ProxyPushConsumer::_nil ();
  64 }
  65
  66 // ****************************************************************
  67
  68 CEC_Counting_Supplier_Task::
  69     CEC_Counting_Supplier_Task (CEC_Counting_Supplier *s,
  70                                 int milliseconds)
  71   :  supplier_ (s),
  72      stop_flag_ (0),
  73      push_count_ (0),
  74      milliseconds_ (milliseconds)
  75 {
  76 }
  77
  78 int
  79 CEC_Counting_Supplier_Task::svc ()
  80 {
  81   try
  82     {
  83       this->run ();
  84     }
  85   catch (const CORBA::Exception&)
  86     {
  87       return -1;
  88     }
  89   return 0;
  90 }
  91
  92 void
  93 CEC_Counting_Supplier_Task::stop ()
  94 {
  95   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
  96   this->stop_flag_ = 1;
  97 }
  98
  99 CORBA::ULong
 100 CEC_Counting_Supplier_Task::push_count ()
 101 {
 102   return this->push_count_;
 103 }
 104
 105 void
 106 CEC_Counting_Supplier_Task::run ()
 107 {
 108   CORBA::Any event;
 109   event <<= CORBA::Long(0);
 110
 111   int stop = 0;
 112   do {
 113     this->supplier_->push (event);
 114
 115     if (this->milliseconds_ != 0)
 116       {
 117         ACE_Time_Value tv (0, 1000 * this->milliseconds_);
 118         ACE_OS::sleep (tv);
 119       }
 120     else
 121       {
 122         // Sleep for a short time to avoid spinning... kind of klugy
 123         ACE_OS::sleep (0);
 124       }
 125
 126     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 127
 128     this->push_count_++;
 129
 130     stop = this->stop_flag_;
 131   } while (stop == 0);
 132 }
 133
 134 // ****************************************************************
 135
 136 CEC_Pull_Counting_Supplier::CEC_Pull_Counting_Supplier ()
 137   :  event_count (0),
 138      disconnect_count (0)
 139 {
 140 }
 141
 142 void
 143 CEC_Pull_Counting_Supplier::connect (
 144     CosEventChannelAdmin::SupplierAdmin_ptr supplier_admin)
 145 {
 146   CosEventComm::PullSupplier_var supplier =
 147     this->_this ();
 148
 149   if (CORBA::is_nil (this->consumer_proxy_.in ()))
 150     {
 151       this->consumer_proxy_ =
 152         supplier_admin->obtain_pull_consumer ();
 153     }
 154
 155   this->consumer_proxy_->connect_pull_supplier (supplier.in ());
 156 }
 157
 158 void
 159 CEC_Pull_Counting_Supplier::disconnect ()
 160 {
 161   if (!CORBA::is_nil (this->consumer_proxy_.in ()))
 162     {
 163       this->consumer_proxy_->disconnect_pull_consumer ();
 164     }
 165
 166   PortableServer::POA_var supplier_poa =
 167     this->_default_POA ();
 168   PortableServer::ObjectId_var supplier_id =
 169     supplier_poa->servant_to_id (this);
 170   supplier_poa->deactivate_object (supplier_id.in ());
 171
 172   this->consumer_proxy_ =
 173     CosEventChannelAdmin::ProxyPullConsumer::_nil ();
 174 }
 175
 176 CORBA::Any*
 177 CEC_Pull_Counting_Supplier::pull ()
 178 {
 179   if (CORBA::is_nil (this->consumer_proxy_.in ()))
 180     throw CosEventComm::Disconnected ();
 181
 182   if (this->event_count % 2)
 183     {
 184       // Sleep for 1 second, to simulate blocking calls.
 185       ACE_OS::sleep (1);
 186     }
 187
 188   this->event_count++;
 189   CORBA::Any event;
 190   event <<= CORBA::Long(0);
 191
 192   return new CORBA::Any (event);
 193 }
 194
 195 CORBA::Any*
 196 CEC_Pull_Counting_Supplier::try_pull (CORBA::Boolean_out has_event)
 197 {
 198   if (CORBA::is_nil (this->consumer_proxy_.in ()))
 199     throw CosEventComm::Disconnected ();
 200
 201   if (this->event_count % 2)
 202     {
 203       // No event
 204       has_event = 0;
 205     }
 206   else
 207     {
 208       // We have an event
 209       has_event = 1;
 210       this->event_count++;
 211    }
 212
 213   CORBA::Any event;
 214   event <<= CORBA::Long(0);
 215
 216   return new CORBA::Any (event);
 217 }
 218
 219 void
 220 CEC_Pull_Counting_Supplier::disconnect_pull_supplier ()
 221 {
 222   this->disconnect_count++;
 223   this->consumer_proxy_ =
 224     CosEventChannelAdmin::ProxyPullConsumer::_nil ();
 225 }