TAO/orbsvcs/tests/CosEvent/Basic/Random.cpp

   1 #include "Random.h"
   2 #include "orbsvcs/CosEvent/CEC_EventChannel.h"
   3 #include "orbsvcs/CosEvent/CEC_Default_Factory.h"
   4 #include "ace/Arg_Shifter.h"
   5 #include "ace/OS_NS_strings.h"
   6 #include "ace/OS_NS_unistd.h"
   7
   8 int
   9 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
  10 {
  11   RND_Driver driver;
  12   return driver.run (argc, argv);
  13 }
  14
  15 // ****************************************************************
  16
  17 void
  18 deactivate_servant (PortableServer::Servant servant)
  19 {
  20   PortableServer::POA_var poa =
  21     servant->_default_POA ();
  22   PortableServer::ObjectId_var oid =
  23     poa->servant_to_id (servant);
  24   poa->deactivate_object (oid.in ());
  25 }
  26
  27
  28 RND_Driver::RND_Driver ()
  29   :  nsuppliers_ (4),
  30      nconsumers_ (4),
  31      max_recursion_ (1)
  32 {
  33   TAO_CEC_Default_Factory::init_svcs ();
  34 }
  35
  36 int
  37 RND_Driver::run (int argc, ACE_TCHAR *argv[])
  38 {
  39   try
  40     {
  41       CORBA::ORB_var orb =
  42         CORBA::ORB_init (argc, argv);
  43
  44       // ****************************************************************
  45
  46       ACE_Arg_Shifter arg_shifter (argc, argv);
  47
  48       while (arg_shifter.is_anything_left ())
  49         {
  50           const ACE_TCHAR *arg = arg_shifter.get_current ();
  51
  52           if (ACE_OS::strcasecmp (arg, ACE_TEXT("-suppliers")) == 0)
  53             {
  54               arg_shifter.consume_arg ();
  55
  56               if (arg_shifter.is_parameter_next ())
  57                 {
  58                   const ACE_TCHAR* opt = arg_shifter.get_current ();
  59                   int n = ACE_OS::atoi (opt);
  60                   if (n >= 1)
  61                     this->nsuppliers_ = n;
  62                   arg_shifter.consume_arg ();
  63                 }
  64             }
  65           else if (ACE_OS::strcasecmp (arg, ACE_TEXT("-consumers")) == 0)
  66             {
  67               arg_shifter.consume_arg ();
  68
  69               if (arg_shifter.is_parameter_next ())
  70                 {
  71                   const ACE_TCHAR* opt = arg_shifter.get_current ();
  72                   int n = ACE_OS::atoi (opt);
  73                   if (n >= 1)
  74                     this->nconsumers_ = n;
  75                   arg_shifter.consume_arg ();
  76                 }
  77             }
  78           else if (ACE_OS::strcasecmp (arg, ACE_TEXT("-max_recursion")) == 0)
  79             {
  80               arg_shifter.consume_arg ();
  81
  82               if (arg_shifter.is_parameter_next ())
  83                 {
  84                   const ACE_TCHAR* opt = arg_shifter.get_current ();
  85                   int n = ACE_OS::atoi (opt);
  86                   if (n >= 0)
  87                     this->max_recursion_ = n;
  88                   arg_shifter.consume_arg ();
  89                 }
  90             }
  91           else
  92             arg_shifter.ignore_arg ();
  93         }
  94
  95       // ****************************************************************
  96
  97       CORBA::Object_var object =
  98         orb->resolve_initial_references ("RootPOA");
  99       PortableServer::POA_var poa =
 100         PortableServer::POA::_narrow (object.in ());
 101       PortableServer::POAManager_var poa_manager =
 102         poa->the_POAManager ();
 103       poa_manager->activate ();
 104
 105       // ****************************************************************
 106
 107       TAO_CEC_EventChannel_Attributes attributes (poa.in (),
 108                                                   poa.in ());
 109       attributes.consumer_reconnect = 1;
 110       attributes.supplier_reconnect = 1;
 111
 112       TAO_CEC_EventChannel ec_impl (attributes);
 113       ec_impl.activate ();
 114
 115       CosEventChannelAdmin::EventChannel_var event_channel =
 116         ec_impl._this ();
 117
 118       // ****************************************************************
 119
 120       // Obtain the consumer admin..
 121       this->consumer_admin_ =
 122         event_channel->for_consumers ();
 123
 124       // Obtain the supplier admin..
 125       this->supplier_admin_ =
 126         event_channel->for_suppliers ();
 127
 128       // ****************************************************************
 129
 130       this->supplier_.connect (this->supplier_admin_.in ());
 131
 132       // ****************************************************************
 133
 134       ACE_NEW_RETURN (this->consumers_,
 135                       RND_Consumer*[this->nconsumers_],
 136                       1);
 137       for (int i = 0; i != this->nconsumers_; ++i)
 138         {
 139           ACE_NEW_RETURN (this->consumers_[i],
 140                           RND_Consumer (this),
 141                           1);
 142
 143           CORBA::Object_var obj =
 144             this->consumers_[i]->_this ();
 145         }
 146
 147       // ****************************************************************
 148
 149       ACE_NEW_RETURN (this->suppliers_,
 150                       RND_Supplier*[this->nsuppliers_],
 151                       1);
 152       for (int j = 0; j != this->nsuppliers_; ++j)
 153         {
 154           ACE_NEW_RETURN (this->suppliers_[j],
 155                           RND_Supplier,
 156                           1);
 157           this->suppliers_[j]->activate ();
 158
 159           CORBA::Object_var obj =
 160             this->suppliers_[j]->_this ();
 161         }
 162
 163       // ****************************************************************
 164
 165       for (int event_count = 0; event_count != 500; ++event_count)
 166         {
 167           ACE_Time_Value tv (0, 50000);
 168           orb->run (tv);
 169           this->supplier_.push_new_event ();
 170         }
 171
 172       ACE_Thread_Manager::instance ()->wait ();
 173
 174       // ****************************************************************
 175
 176       {
 177         for (int k = 0; k != this->nsuppliers_; ++k)
 178           {
 179             deactivate_servant (this->suppliers_[k]);
 180             this->suppliers_[k]->_remove_ref ();
 181           }
 182         delete[] this->suppliers_;
 183         this->suppliers_ = 0;
 184       }
 185
 186       // ****************************************************************
 187
 188       // We destroy now to verify that the callbacks work and do not
 189       // produce any problems.
 190       event_channel->destroy ();
 191
 192       // ****************************************************************
 193
 194       {
 195         for (int k = 0; k != this->nconsumers_; ++k)
 196           {
 197             deactivate_servant (this->consumers_[k]);
 198             this->consumers_[k]->_remove_ref ();
 199           }
 200         delete[] this->consumers_;
 201         this->consumers_ = 0;
 202       }
 203
 204       // ****************************************************************
 205
 206       deactivate_servant (&ec_impl);
 207
 208       // ****************************************************************
 209
 210       poa->destroy (true, true);
 211
 212       // ****************************************************************
 213
 214       orb->destroy ();
 215     }
 216   catch (const CORBA::Exception& ex)
 217     {
 218       ex._tao_print_exception ("Random");
 219       return 1;
 220     }
 221   return 0;
 222 }
 223
 224 void
 225 RND_Driver::timer (const CORBA::Any &e)
 226 {
 227   int r = ACE_OS::rand ();
 228   if (r < 0)
 229     r = -r;
 230
 231   int n = r% 20;
 232
 233   switch (n)
 234     {
 235     case 0:
 236     case 1:
 237       {
 238         CORBA::Long recursion;
 239         e >>= recursion;
 240         // ACE_DEBUG ((LM_DEBUG, "Pushing an event\n"));
 241         if (recursion < this->max_recursion_)
 242           {
 243             CORBA::Any new_event;
 244             recursion++;
 245             new_event <<= recursion;
 246             this->supplier_.push (new_event);
 247           }
 248       }
 249       break;
 250
 251     default:
 252     case 2:
 253     case 3:
 254     case 4:
 255     case 5:
 256       // ACE_DEBUG ((LM_DEBUG, "Received event\n"));
 257       break;
 258
 259     case 6:
 260       {
 261         int n = ACE_OS::rand () % this->nsuppliers_;
 262
 263         // ACE_DEBUG ((LM_DEBUG, "Connecting supplier %d\n", n));
 264
 265         this->suppliers_[n]->connect (this->supplier_admin_.in ());
 266       }
 267       break;
 268
 269     case 7:
 270       {
 271         int n = ACE_OS::rand () % this->nconsumers_;
 272
 273         // ACE_DEBUG ((LM_DEBUG, "Connecting consumer %d\n", n));
 274
 275         this->consumers_[n]->connect (this->consumer_admin_.in ());
 276       }
 277       break;
 278
 279     case 8:
 280       {
 281         int n = ACE_OS::rand () % this->nsuppliers_;
 282
 283         // ACE_DEBUG ((LM_DEBUG, "Disconnecting supplier %d\n", n));
 284
 285         this->suppliers_[n]->disconnect ();
 286       }
 287       break;
 288
 289     case 9:
 290       {
 291         int n = ACE_OS::rand () % this->nconsumers_;
 292
 293         // ACE_DEBUG ((LM_DEBUG, "Disconnecting consumer %d\n", n));
 294
 295         this->consumers_[n]->disconnect ();
 296       }
 297       break;
 298     }
 299 }
 300
 301 void
 302 RND_Driver::event (const CORBA::Any &e)
 303 {
 304   this->timer (e);
 305 }
 306
 307 // ****************************************************************
 308
 309 void
 310 RND_Consumer::connect (CosEventChannelAdmin::ConsumerAdmin_ptr admin)
 311 {
 312   CosEventChannelAdmin::ProxyPushSupplier_var proxy;
 313   {
 314     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 315
 316     if (CORBA::is_nil (this->proxy_.in ()))
 317       {
 318         this->proxy_ = admin->obtain_push_supplier ();
 319       }
 320     proxy =
 321       CosEventChannelAdmin::ProxyPushSupplier::_duplicate(this->proxy_.in ());
 322   }
 323   CosEventComm::PushConsumer_var me =
 324     this->_this ();
 325   proxy->connect_push_consumer (me.in ());
 326 }
 327
 328 void
 329 RND_Consumer::disconnect ()
 330 {
 331   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 332
 333   if (CORBA::is_nil (this->proxy_.in ()))
 334     return;
 335   this->proxy_->disconnect_push_supplier ();
 336   this->proxy_ =
 337     CosEventChannelAdmin::ProxyPushSupplier::_nil ();
 338 }
 339
 340 void
 341 RND_Consumer::push (const CORBA::Any &event)
 342 {
 343   this->driver_->event (event);
 344 }
 345
 346 void
 347 RND_Consumer::disconnect_push_consumer ()
 348 {
 349 }
 350
 351 // ****************************************************************
 352
 353 void
 354 RND_Supplier::connect (CosEventChannelAdmin::SupplierAdmin_ptr admin)
 355 {
 356   CosEventChannelAdmin::ProxyPushConsumer_var proxy;
 357   {
 358     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 359
 360     if (CORBA::is_nil (this->proxy_.in ()))
 361       {
 362         this->proxy_ = admin->obtain_push_consumer ();
 363       }
 364
 365     proxy =
 366       CosEventChannelAdmin::ProxyPushConsumer::_duplicate(this->proxy_.in ());
 367   }
 368   CosEventComm::PushSupplier_var me;
 369
 370   int r = ACE_OS::rand () % 2;
 371   if (r == 0)
 372     {
 373       me = this->_this ();
 374     }
 375   proxy->connect_push_supplier (me.in ());
 376 }
 377
 378 void
 379 RND_Supplier::disconnect ()
 380 {
 381   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 382
 383   if (CORBA::is_nil (this->proxy_.in ()))
 384     return;
 385   this->proxy_->disconnect_push_consumer ();
 386   this->proxy_ =
 387     CosEventChannelAdmin::ProxyPushConsumer::_nil ();
 388 }
 389
 390 void
 391 RND_Supplier::push_new_event ()
 392 {
 393   CORBA::Any event;
 394   CORBA::Long recursion = 0;
 395   event <<= recursion;
 396
 397   this->push (event);
 398 }
 399
 400 void
 401 RND_Supplier::push (CORBA::Any &event)
 402 {
 403   CosEventChannelAdmin::ProxyPushConsumer_var proxy;
 404   {
 405     ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 406
 407     if (CORBA::is_nil (this->proxy_.in ()))
 408       return;
 409
 410     proxy =
 411       CosEventChannelAdmin::ProxyPushConsumer::_duplicate(this->proxy_.in ());
 412   }
 413
 414   proxy->push (event);
 415 }
 416
 417 void
 418 RND_Supplier::disconnect_push_supplier ()
 419 {
 420 }
 421
 422 int
 423 RND_Supplier::svc ()
 424 {
 425   ACE_DEBUG ((LM_DEBUG, "Thread %t started\n"));
 426   int percent = 10;
 427   int niterations = 5000;
 428   for (int i = 0; i != niterations; ++i)
 429     {
 430       try
 431         {
 432           ACE_Time_Value tv (0, 10000);
 433           ACE_OS::sleep (tv);
 434
 435           this->push_new_event ();
 436         }
 437       catch (const CORBA::Exception&)
 438         {
 439         }
 440       if (i * 100 / niterations >= percent)
 441         {
 442           ACE_DEBUG ((LM_DEBUG, "Thread %t %d%%\n", percent));
 443           percent += 10;
 444         }
 445     }
 446   ACE_DEBUG ((LM_DEBUG, "Thread %t completed\n"));
 447   return 0;
 448 }