TAO/orbsvcs/tests/Notify/Basic/AdminProperties.cpp

   1 #include "AdminProperties.h"
   2 #include "ace/Arg_Shifter.h"
   3 #include "ace/Get_Opt.h"
   4 #include "ace/OS_NS_unistd.h"
   5 #include "tao/debug.h"
   6
   7 /***************************************************************************/
   8
   9 AdminProperties_Task::AdminProperties_Task ()
  10   : supplier_ (0)
  11   , client_ (0)
  12 {
  13 }
  14
  15 void
  16 AdminProperties_Task::init (TAO_Notify_Tests_StructuredPushSupplier *supplier, AdminProperties* client)
  17 {
  18   supplier_  = supplier;
  19   client_ = client;
  20 }
  21
  22 int
  23 AdminProperties_Task::init (int argc, ACE_TCHAR *argv[])
  24 {
  25   return ACE_Task_Base::init (argc, argv);
  26 }
  27
  28 int
  29 AdminProperties_Task::svc ()
  30 {
  31   // operations:
  32   CosNotification::StructuredEvent event;
  33
  34   // EventHeader
  35
  36   // FixedEventHeader
  37   // EventType
  38   // string
  39   event.header.fixed_header.event_type.domain_name = CORBA::string_dup("*");
  40   // string
  41   event.header.fixed_header.event_type.type_name = CORBA::string_dup("*");
  42   // string
  43   event.header.fixed_header.event_name = CORBA::string_dup("myevent");
  44
  45   // OptionalHeaderFields
  46   // PropertySeq
  47   // sequence<Property>: string name, any value
  48   event.header.variable_header.length (0); // put nothing here
  49
  50   // FilterableEventBody
  51   // PropertySeq
  52   // sequence<Property>: string name, any value
  53   event.filterable_data.length (3);
  54   event.filterable_data[0].name = CORBA::string_dup("threshold");
  55
  56   event.filterable_data[1].name = CORBA::string_dup("temperature");
  57   event.filterable_data[1].value <<= (CORBA::Long)70;
  58
  59   event.filterable_data[2].name = CORBA::string_dup("pressure");
  60   event.filterable_data[2].value <<= (CORBA::Long)80;
  61
  62   // @@ CORBA::Short prio = CosNotification::LowestPriority;
  63
  64  int event_count = this->client_->event_count_;
  65
  66  ACE_DEBUG ((LM_DEBUG, "\n1 supplier sending %d events...\n", event_count));
  67
  68
  69  for (int i = 0 ; i < event_count; ++i)
  70    {
  71      event.filterable_data[0].value <<= (CORBA::Long)i;
  72
  73      event.remainder_of_body <<= (CORBA::Long)i;
  74
  75      try
  76         {
  77           ACE_DEBUG((LM_DEBUG, "+"));
  78           this->supplier_->send_event (event);
  79         }
  80       catch (const CORBA::IMP_LIMIT&)
  81         {
  82           if (TAO_debug_level > 0)
  83             ACE_DEBUG ((LM_DEBUG, "\nEvent %d was not send due to Impl Limit reached\n", i));
  84
  85           ++ this->client_->rejections_;
  86         }
  87       catch (const CORBA::Exception& ex)
  88         {
  89           ex._tao_print_exception (
  90             "Error: Exception sending event\n");
  91           return 1;
  92         }
  93    }
  94
  95 return 0;
  96 }
  97 /***************************************************************************/
  98
  99 AdminProperties_StructuredPushConsumer::AdminProperties_StructuredPushConsumer (AdminProperties* client)
 100   : client_ (client)
 101   , events_received_ (0)
 102 {
 103   client_->consumer_start(this);
 104 }
 105
 106 void
 107 AdminProperties_StructuredPushConsumer::push_structured_event (const CosNotification::StructuredEvent & /*notification*/
 108                                                                )
 109 {
 110   ++events_received_;
 111
 112   if (events_received_ >= client_->max_queue_length_)
 113     client_->consumer_done(this);
 114
 115   ACE_DEBUG((LM_DEBUG, "-"));
 116
 117   if (TAO_debug_level)
 118     ACE_DEBUG ((LM_DEBUG, "\nConsumer %x received event %d\n", this, events_received_.value ()));
 119 }
 120
 121 /***************************************************************************/
 122
 123 AdminProperties::AdminProperties ()
 124   : max_queue_length_ (10),
 125     max_consumers_ (3),
 126     max_suppliers_ (3),
 127     reject_new_events_ (0),
 128     consumers_ (4),
 129     suppliers_ (4),
 130     event_count_ (30),
 131     suppliers_connected_count_ (0),
 132     consumers_connected_count_ (0),
 133     rejections_ (0)
 134 {
 135 }
 136
 137 AdminProperties::~AdminProperties ()
 138 {
 139 }
 140
 141 int
 142 AdminProperties::parse_args(int argc, ACE_TCHAR *argv[])
 143 {
 144   ACE_Arg_Shifter arg_shifter (argc, argv);
 145
 146   const ACE_TCHAR *current_arg = 0;
 147
 148   while (arg_shifter.is_anything_left ())
 149     {
 150       if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-max_queue_length"))))
 151         {
 152           this->max_queue_length_ = ACE_OS::atoi (current_arg);
 153           // Max. queue length.
 154
 155           arg_shifter.consume_arg ();
 156         }
 157       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-max_consumers"))))
 158         {
 159           this->max_consumers_ = ACE_OS::atoi (current_arg);
 160           // Max consumers allowed to connect.
 161           arg_shifter.consume_arg ();
 162         }
 163       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-max_suppliers"))))
 164         {
 165           this->max_suppliers_ = ACE_OS::atoi (current_arg);
 166           // Max. number of suppliers allowed to connect.
 167           arg_shifter.consume_arg ();
 168         }
 169       else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-reject_new_events")) == 0)
 170         {
 171           this->reject_new_events_ = 1;
 172           arg_shifter.consume_arg ();
 173         }
 174       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-consumers"))))
 175         {
 176           this->consumers_ = ACE_OS::atoi (current_arg);
 177           // Number of consumers to create.
 178           arg_shifter.consume_arg ();
 179         }
 180       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-suppliers"))))
 181         {
 182           this->suppliers_ = ACE_OS::atoi (current_arg);
 183           // Number of suppliers to create.
 184           arg_shifter.consume_arg ();
 185         }
 186       else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-?")) == 0)
 187         {
 188           ACE_DEBUG((LM_DEBUG,
 189                      "usage: %s "
 190                      "-max_queue_length [max_queue_length] "
 191                      "-max_consumers [max_consumers] "
 192                      "-max_suppliers [max_suppliers] "
 193                      "-reject_new_events [reject_new_events] "
 194                      "-consumers [consumers] "
 195                      "-suppliers [suppliers] "
 196                      "-event_count [event_count] ",
 197                      argv[0],
 198                      argv[0]));
 199
 200           arg_shifter.consume_arg ();
 201
 202           return -1;
 203         }
 204       else
 205         {
 206           arg_shifter.ignore_arg ();
 207         }
 208     }
 209     return 0;
 210 }
 211
 212 void
 213 AdminProperties::create_channel(bool reject)
 214 {
 215   CosNotifyChannelAdmin::ChannelID id;
 216
 217   initial_admin_.length (4);
 218
 219   this->initial_admin_[0].name =
 220     CORBA::string_dup (CosNotification::MaxQueueLength);
 221   this->initial_admin_[0].value <<= this->max_queue_length_;
 222
 223
 224   this->initial_admin_[1].name =
 225     CORBA::string_dup (CosNotification::MaxSuppliers);
 226   this->initial_admin_[1].value <<= this->max_suppliers_;
 227
 228   this->initial_admin_[2].name =
 229     CORBA::string_dup (CosNotification::MaxConsumers);
 230   this->initial_admin_[2].value <<= this->max_consumers_;
 231
 232
 233   this->initial_admin_[3].name =
 234     CORBA::string_dup (CosNotification::RejectNewEvents);
 235   this->initial_admin_[3].value <<= CORBA::Any::from_boolean (reject);
 236
 237   this->ec_ = notify_factory_->create_channel (this->initial_qos_,
 238                                                this->initial_admin_,
 239                                                id);
 240
 241   ACE_ASSERT (!CORBA::is_nil (ec_.in ()));
 242
 243
 244   CosNotifyChannelAdmin::AdminID adminid;
 245
 246   this->supplier_admin_ = ec_->new_for_suppliers (this->ifgop_,
 247                                                   adminid);
 248
 249   ACE_ASSERT (!CORBA::is_nil (supplier_admin_.in ()));
 250
 251   this->consumer_admin_ = ec_->new_for_consumers (this->ifgop_,
 252                                                   adminid);
 253
 254   ACE_ASSERT (!CORBA::is_nil (consumer_admin_.in ()));
 255 }
 256
 257 void
 258 AdminProperties::run_test ()
 259 {
 260   bool reject = true;
 261   this->create_channel(reject);
 262   this->test_max_queue_length (reject);
 263
 264   this->ec_->destroy();
 265
 266   reject = false;
 267   this->create_channel(reject);
 268   this->test_max_queue_length (reject);
 269
 270    this->test_max_clients ();
 271
 272   this->ec_->destroy ();
 273 }
 274
 275 void
 276 AdminProperties::test_max_queue_length (bool reject)
 277 {
 278   // Create the consumer
 279   AdminProperties_StructuredPushConsumer *consumer;
 280   ACE_NEW (consumer, AdminProperties_StructuredPushConsumer (this));
 281   consumer->init (root_poa_.in ());
 282   consumer->connect (this->consumer_admin_.in ());
 283
 284   // Create the supplier
 285   TAO_Notify_Tests_StructuredPushSupplier *supplier = 0;
 286   ACE_NEW (supplier, TAO_Notify_Tests_StructuredPushSupplier ());
 287   supplier->init (root_poa_.in ());
 288   supplier->connect (this->supplier_admin_.in ());
 289
 290   AdminProperties_Task supplier_task;
 291
 292   // Init the Task to send events;
 293   supplier_task.init (supplier, this);
 294
 295   if (supplier_task.activate (THR_NEW_LWP | THR_JOINABLE, 1) != 0)
 296     {
 297     ACE_ERROR ((LM_ERROR, "\nCannot activate supplier task\n"));
 298     }
 299
 300   // All supplier events should be sent before the first consumer event is
 301   // received. This relies on our use of -ORBClientConnectionHandler RW.
 302   supplier_task.wait ();
 303
 304   this->ORB_run();
 305
 306   // consumer is destroyed by consumer->disconnect()
 307   CORBA::Long received_count = consumer->events_received_.value ();
 308
 309   // disconnect the participants.
 310   consumer->disconnect ();
 311   supplier->disconnect ();
 312
 313   // If the reject_new_events setting == true, then the supplier should
 314   // have received an imp_limit exception for each event it tried to push
 315   // after the maximum was reached.
 316   // If the reject_new_events setting == false, then the events should
 317   // have been discarded according to the DiscardPolicy, which for this
 318   // test we leave as AnyOrder.
 319
 320   ACE_DEBUG ((LM_DEBUG, "\nSupplier sent %d events, consumer received %d events, max_queue_length = %d\n",
 321               event_count_, received_count, max_queue_length_));
 322
 323   int expected_min = max_queue_length_;
 324   int expected_max = max_queue_length_ + max_consumers_;
 325   if (reject)
 326   {
 327     expected_max = event_count_ - rejections_;
 328     expected_min = expected_max;
 329   }
 330
 331   if (reject && rejections_ != event_count_ - received_count)
 332     {
 333     ACE_ERROR ((LM_ERROR, "\nError: Expected %d rejections, but got %d\n",
 334       event_count_ - received_count, rejections_));
 335     return;
 336   }
 337
 338   if (received_count < expected_min || received_count > expected_max)
 339   {
 340     ACE_ERROR ((LM_ERROR, "\nError: Expected %d to %d events, but received %d\n",
 341       expected_min, expected_max, received_count));
 342     }
 343 }
 344
 345 void
 346 AdminProperties::test_max_clients ()
 347 {
 348   this->create_suppliers ();
 349
 350   this->create_consumers ();
 351
 352   // check the results and print the assessment.
 353   if (this->consumers_connected_count_ > this->max_consumers_)
 354     ACE_DEBUG ((LM_ERROR, "\nConnected consumers %d, exceed MaxConsumers %d\n",
 355                 this->consumers_connected_count_ > this->max_consumers_));
 356
 357   if (this->suppliers_connected_count_ > this->max_suppliers_)
 358     ACE_DEBUG ((LM_ERROR, "\nConnected suppliers %d, exceed MaxSuppliers %d\n",
 359                 this->suppliers_connected_count_ > this->max_suppliers_));
 360 }
 361
 362 void
 363 AdminProperties::create_suppliers ()
 364 {
 365   // Create the requested number of suppliers.
 366   // @@ CosNotifyChannelAdmin::AdminID adminid;
 367   // @@ CosNotifyChannelAdmin::InterFilterGroupOperator ifgop =
 368   // @@   CosNotifyChannelAdmin::OR_OP;
 369
 370   int index = 0;
 371
 372   try
 373     {
 374       TAO_Notify_Tests_StructuredPushSupplier *supplier;
 375
 376       for (index = 0; index < this->suppliers_; ++index)
 377         {
 378           ACE_NEW (supplier,
 379                    TAO_Notify_Tests_StructuredPushSupplier ());
 380           supplier->init (root_poa_.in ());
 381
 382           supplier->connect (this->supplier_admin_.in ());
 383
 384           this->suppliers_connected_count_++;
 385         }
 386     }
 387   catch (const CORBA::IMP_LIMIT&)
 388     {
 389       if (TAO_debug_level)
 390         ACE_DEBUG ((LM_DEBUG, "\nImpl Limit excpetion when connecting supplier\n"));
 391     }
 392   catch (const CORBA::Exception& ex)
 393     {
 394       ex._tao_print_exception ("\nError: Exception in connecting supplier\n");
 395     }
 396 }
 397
 398 void
 399 AdminProperties::create_consumers ()
 400 {
 401   // Create the requested number of suppliers.
 402   // @@ CosNotifyChannelAdmin::AdminID adminid;
 403   // @@ CosNotifyChannelAdmin::InterFilterGroupOperator ifgop =
 404   // @@   CosNotifyChannelAdmin::OR_OP;
 405
 406   int index = 0;
 407
 408   try
 409     {
 410       TAO_Notify_Tests_StructuredPushConsumer *consumer;
 411
 412       for (index = 0; index < this->consumers_; ++index)
 413         {
 414           ACE_NEW (consumer, TAO_Notify_Tests_StructuredPushConsumer ());
 415           consumer->init (root_poa_.in ());
 416
 417           consumer->connect (this->consumer_admin_.in ());
 418
 419           this->consumers_connected_count_++;
 420         }
 421     }
 422   catch (const CORBA::IMP_LIMIT&)
 423     {
 424       if (TAO_debug_level)
 425         ACE_DEBUG ((LM_DEBUG, "\nImpl Limit exception when connecting consumer\n"));
 426     }
 427   catch (const CORBA::Exception& ex)
 428     {
 429         ex._tao_print_exception (
 430           "\nError: Exception in connecting consumer\n");
 431     }
 432 }
 433
 434 /***************************************************************************/
 435
 436 int
 437 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
 438 {
 439   AdminProperties test;
 440
 441   try
 442     {
 443       test.init (argc, argv);
 444
 445       test.run_test ();
 446     }
 447   catch (const CORBA::Exception& se)
 448     {
 449       se._tao_print_exception ("Error: ");
 450       return 1;
 451     }
 452   return 0;
 453 }