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

   1 #include "ace/Arg_Shifter.h"
   2 #include "ace/Get_Opt.h"
   3 #include "ace/OS_NS_unistd.h"
   4 #include "tao/debug.h"
   5 #include "Sequence.h"
   6
   7
   8 /***************************************************************************/
   9
  10 SequencePushConsumer::SequencePushConsumer (Sequence *test_client)
  11   : test_client_ (test_client)
  12 {
  13 }
  14
  15 void
  16 SequencePushConsumer::push_structured_events (
  17     const CosNotification::EventBatch &batch)
  18 {
  19   this->test_client_->events_received_ += batch.length ();
  20
  21   if (batch.length () > this->test_client_->consumer_batch_size_)
  22       ACE_DEBUG ((LM_ERROR,
  23                   "Error: Received more than max event batch %d\n",
  24                   batch.length ()));
  25
  26   this->test_client_->on_event_received ();
  27
  28   ACE_OS::sleep (this->test_client_->consumer_delay_);
  29 }
  30
  31 /***************************************************************************/
  32
  33 SequencePushSupplier::SequencePushSupplier (
  34     Sequence* test_client)
  35   : test_client_ (test_client)
  36 {
  37 }
  38
  39 SequencePushSupplier::~SequencePushSupplier ()
  40 {
  41 }
  42
  43 /***************************************************************************/
  44 Sequence::Sequence ()
  45   : event_count_ (15), supplier_batch_size_ (5), consumer_batch_size_ (3),
  46     pacing_ (2), order_policy_ (CosNotification::PriorityOrder), events_received_ (0),
  47     consumer_delay_ (1)
  48 {
  49 }
  50
  51 Sequence::~Sequence ()
  52 {
  53 }
  54
  55 int
  56 Sequence::init (int argc,
  57                    ACE_TCHAR* argv [])
  58 {
  59   if (TAO_debug_level)
  60     {
  61       ACE_DEBUG ((LM_DEBUG,
  62                   "Options: event count = %d\n"
  63                   "supplier batch size = %d\n"
  64                   "consumer batch size = %d\n"
  65                   "pacing = %d secs\n"
  66                   , event_count_
  67                   , supplier_batch_size_
  68                   , consumer_batch_size_
  69                   , pacing_));
  70
  71       ACE_DEBUG ((LM_DEBUG, "consumer delay = %d\n", consumer_delay_.sec ()));
  72     }
  73
  74   // Initialize the base class.
  75   Notify_Test_Client::init (argc,
  76                             argv);
  77
  78   // Create all participents.
  79   this->create_EC ();
  80
  81   CosNotifyChannelAdmin::AdminID adminid;
  82
  83   this->supplier_admin_ =
  84     this->ec_->new_for_suppliers (this->ifgop_,
  85                                   adminid);
  86
  87   ACE_ASSERT (!CORBA::is_nil (supplier_admin_.in ()));
  88
  89   this->consumer_admin_ =
  90     this->ec_->new_for_consumers (this->ifgop_,
  91                                   adminid);
  92
  93   ACE_ASSERT (!CORBA::is_nil (consumer_admin_.in ()));
  94
  95   ACE_NEW_RETURN (this->consumer_,
  96                   SequencePushConsumer (this),
  97                   -1);
  98   this->consumer_->init (root_poa_.in ());
  99   this->consumer_->connect (this->consumer_admin_.in ());
 100
 101   CosNotification::QoSProperties properties (3);
 102   properties.length (3);
 103
 104   properties[0].name = CORBA::string_dup (CosNotification::MaximumBatchSize);
 105   properties[0].value <<= (CORBA::Long) this->consumer_batch_size_;
 106   properties[1].name = CORBA::string_dup (CosNotification::PacingInterval);
 107   properties[1].value <<= (TimeBase::TimeT) (this->pacing_ * 1000 * 10000);
 108   properties[2].name = CORBA::string_dup (CosNotification::OrderPolicy);
 109   properties[2].value <<= this->order_policy_;
 110
 111   this->consumer_->get_proxy_supplier ()->set_qos (properties);
 112
 113   ACE_NEW_RETURN (this->supplier_,
 114                   SequencePushSupplier (this),
 115                   -1);
 116   this->supplier_->init (root_poa_.in ());
 117
 118   this->supplier_->connect (this->supplier_admin_.in ());
 119
 120   consumer_start (0);
 121
 122   return 0;
 123 }
 124
 125 int
 126 Sequence::parse_args (int argc,
 127                          ACE_TCHAR *argv[])
 128 {
 129     ACE_Arg_Shifter arg_shifter (argc,
 130                                  argv);
 131     const ACE_TCHAR *current_arg = 0;
 132
 133     while (arg_shifter.is_anything_left ())
 134     {
 135       if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-events"))))
 136         {
 137           this->event_count_ = ACE_OS::atoi (current_arg); // The number of events to send/receive.
 138
 139           arg_shifter.consume_arg ();
 140         }
 141       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-SupplierBatchSize"))))
 142         {
 143           this->supplier_batch_size_ = ACE_OS::atoi (current_arg); // Supplier batch size
 144
 145           arg_shifter.consume_arg ();
 146         }
 147       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-ConsumerBatchSize"))))
 148         {
 149           this->consumer_batch_size_ = ACE_OS::atoi (current_arg); // Consumer batch size
 150
 151           arg_shifter.consume_arg ();
 152         }
 153       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-ConsumerDelay"))))
 154         {
 155           this->consumer_delay_ = ACE_Time_Value (ACE_OS::atoi (current_arg), 0); // Consumer delay in secs.
 156
 157           arg_shifter.consume_arg ();
 158         }
 159       else if (0 != (current_arg = arg_shifter.get_the_parameter (ACE_TEXT("-Pacing")))) // in seconds
 160         {
 161           this->pacing_ = (TimeBase::TimeT) ACE_OS::atoi (current_arg);
 162
 163           arg_shifter.consume_arg ();
 164         }
 165
 166       else if (arg_shifter.cur_arg_strncasecmp (ACE_TEXT("-?")) == 0)
 167         {
 168           ACE_DEBUG((LM_DEBUG,
 169                      "usage: %s "
 170                      "-events event_count "
 171                      "-SupplierBatchSize size "
 172                      "-ConsumerBatchSize size "
 173                      "-ConsumerDelay delay "
 174                      "-Pacing pacing\n",
 175                      argv[0], argv[0]));
 176
 177           arg_shifter.consume_arg ();
 178
 179           return -1;
 180         }
 181       else
 182         {
 183           arg_shifter.ignore_arg ();
 184         }
 185     }
 186
 187   return 0;
 188 }
 189
 190 void
 191 Sequence::create_EC ()
 192 {
 193   CosNotifyChannelAdmin::ChannelID id;
 194
 195   this->ec_ = notify_factory_->create_channel (this->initial_qos_,
 196                                                this->initial_admin_,
 197                                                id);
 198   ACE_ASSERT (!CORBA::is_nil (this->ec_.in ()));
 199 }
 200
 201 void
 202 Sequence::on_event_received ()
 203 {
 204   if (TAO_debug_level)
 205     ACE_DEBUG ((LM_DEBUG,
 206                 "Events received = %d\n",
 207                 this->events_received_.value ()));
 208
 209   if (this->events_received_.value () == this->event_count_)
 210     {
 211       this->end_test ();
 212     }
 213 }
 214
 215 void
 216 Sequence::run_test ()
 217 {
 218   // operations:
 219   CosNotification::StructuredEvent event;
 220
 221   // EventHeader.
 222
 223   // FixedEventHeader.
 224   // EventType.
 225   // string.
 226   event.header.fixed_header.event_type.domain_name = CORBA::string_dup("*");
 227   // string
 228   event.header.fixed_header.event_type.type_name = CORBA::string_dup("*");
 229   // string
 230   event.header.fixed_header.event_name = CORBA::string_dup("myevent");
 231
 232   // OptionalHeaderFields.
 233   // PropertySeq.
 234   // sequence<Property>: string name, any value
 235   CosNotification::PropertySeq& qos =  event.header.variable_header;
 236   qos.length (1); // put nothing here
 237
 238   // FilterableEventBody
 239   // PropertySeq
 240   // sequence<Property>: string name, any value
 241   event.filterable_data.length (3);
 242   event.filterable_data[0].name = CORBA::string_dup("threshold");
 243
 244   event.filterable_data[1].name = CORBA::string_dup("temperature");
 245   event.filterable_data[1].value <<= (CORBA::Long)70;
 246
 247   event.filterable_data[2].name = CORBA::string_dup("pressure");
 248   event.filterable_data[2].value <<= (CORBA::Long)80;
 249
 250   CORBA::Short prio = CosNotification::LowestPriority;
 251
 252   CosNotification::EventBatch batch;
 253   batch.length (this->supplier_batch_size_);
 254   CORBA::ULong batch_index = 0;
 255
 256   for (int i = 0; i < this->event_count_; ++i)
 257     {
 258       event.filterable_data[0].value <<= (CORBA::Long)i;
 259
 260       // any
 261       event.remainder_of_body <<= (CORBA::Long)i;
 262
 263       qos[0].name = CORBA::string_dup (CosNotification::Priority);
 264       qos[0].value <<= (CORBA::Short)prio++;
 265
 266       batch[batch_index] = event;
 267       batch_index++;
 268
 269       if (batch_index == this->supplier_batch_size_)
 270         {
 271           batch.length (batch_index); // set the correct length
 272
 273           if (TAO_debug_level)
 274             ACE_DEBUG ((LM_DEBUG, "Sending batch with %d events\n", batch.length ()));
 275
 276           this->supplier_->send_events (batch);
 277
 278           // reset
 279           batch.length (this->supplier_batch_size_);
 280           batch_index = 0;
 281         }
 282     } // for
 283
 284   // send the last batch.
 285   if (batch_index > 0)
 286     {
 287       batch.length (batch_index); // set the correct length
 288
 289       this->supplier_->send_events (batch);
 290     }
 291
 292 }
 293
 294 void
 295 Sequence::end_test ()
 296 {
 297   consumer_done(0);
 298 }
 299
 300 int
 301 Sequence::check_results ()
 302 {
 303   // Destroy the channel.
 304   this->ec_->destroy ();
 305
 306   if (this->events_received_.value () == this->event_count_)
 307     {
 308       ACE_DEBUG ((LM_DEBUG,
 309                   "Sequence test success\n"));
 310       return 0;
 311     }
 312   else
 313     {
 314       ACE_DEBUG ((LM_DEBUG,
 315                   "Sequence test failed!\n"));
 316       return 1;
 317     }
 318 }
 319
 320 /***************************************************************************/
 321
 322 int
 323 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
 324 {
 325   Sequence events;
 326
 327   if (events.parse_args (argc, argv) == -1)
 328     {
 329       return 1;
 330     }
 331
 332   try
 333     {
 334       events.init (argc, argv);
 335
 336       events.run_test ();
 337
 338       events.ORB_run();
 339     }
 340   catch (const CORBA::Exception& se)
 341     {
 342       se._tao_print_exception ("Error: ");
 343       return 1;
 344     }
 345
 346   int status = 0;
 347
 348   try
 349     {
 350       status = events.check_results ();
 351     }
 352   catch (const CORBA::Exception& se)
 353     {
 354       se._tao_print_exception ("Error: ");
 355       status = 1;
 356     }
 357
 358   return status;
 359 }