TAO/orbsvcs/tests/EC_Custom_Marshal/ECM_Consumer.cpp

   1 #include "ECM_Consumer.h"
   2 #include "ECM_Data.h"
   3
   4 #include "orbsvcs/Event_Utilities.h"
   5 #include "orbsvcs/Event_Service_Constants.h"
   6 #include "orbsvcs/Time_Utilities.h"
   7 #include "orbsvcs/CosNamingC.h"
   8
   9 #include "tao/Timeprobe.h"
  10 #include "tao/ORB_Core.h"
  11 #include "tao/CDR.h"
  12
  13 #include "ace/Get_Opt.h"
  14 #include "ace/Auto_Ptr.h"
  15 #include "ace/Sched_Params.h"
  16 #include "ace/OS_NS_errno.h"
  17 #include "ace/OS_NS_unistd.h"
  18
  19 int
  20 ACE_TMAIN(int argc, ACE_TCHAR *argv[])
  21 {
  22   Driver driver;
  23   return driver.run (argc, argv);
  24 }
  25
  26 // ****************************************************************
  27
  28 Driver::Driver (void)
  29   : n_consumers_ (1),
  30     event_count_ (100),
  31     event_a_ (ACE_ES_EVENT_UNDEFINED),
  32     event_b_ (ACE_ES_EVENT_UNDEFINED + 1),
  33     pid_file_name_ (0),
  34     recv_count_ (0)
  35 {
  36 }
  37
  38 // ****************************************************************
  39
  40 int
  41 Driver::run (int argc, ACE_TCHAR* argv[])
  42 {
  43   try
  44     {
  45       CORBA::ORB_var orb =
  46         CORBA::ORB_init (argc, argv);
  47
  48       CORBA::Object_var poa_object =
  49         orb->resolve_initial_references("RootPOA");
  50
  51       if (CORBA::is_nil (poa_object.in ()))
  52         ACE_ERROR_RETURN ((LM_ERROR,
  53                            " (%P|%t) Unable to initialize the POA.\n"),
  54                           1);
  55
  56       PortableServer::POA_var root_poa =
  57         PortableServer::POA::_narrow (poa_object.in ());
  58
  59       PortableServer::POAManager_var poa_manager =
  60         root_poa->the_POAManager ();
  61
  62       if (this->parse_args (argc, argv))
  63         return 1;
  64
  65       ACE_DEBUG ((LM_DEBUG,
  66                   "Execution parameters:\n"
  67                   "  consumers = <%d>\n"
  68                   "  event count = <%d>\n"
  69                   "  supplier Event A = <%d>\n"
  70                   "  supplier Event B = <%d>\n"
  71                   "  pid file name = <%s>\n",
  72
  73                   this->n_consumers_,
  74                   this->event_count_,
  75                   this->event_a_,
  76                   this->event_b_,
  77
  78                   this->pid_file_name_?this->pid_file_name_:ACE_TEXT("nil")) );
  79
  80       if (this->pid_file_name_ != 0)
  81         {
  82           FILE* pid = ACE_OS::fopen (this->pid_file_name_, "w");
  83           if (pid != 0)
  84             {
  85               ACE_OS::fprintf (pid, "%ld\n",
  86                                static_cast<long> (ACE_OS::getpid ()));
  87               ACE_OS::fclose (pid);
  88             }
  89         }
  90
  91       int min_priority =
  92         ACE_Sched_Params::priority_min (ACE_SCHED_FIFO);
  93         // Enable FIFO scheduling, e.g., RT scheduling class on Solaris.
  94
  95       if (ACE_OS::sched_params (ACE_Sched_Params (ACE_SCHED_FIFO,
  96                                                   min_priority,
  97                                                   ACE_SCOPE_PROCESS)) != 0)
  98         {
  99           if (ACE_OS::last_error () == EPERM)
 100             ACE_DEBUG ((LM_DEBUG,
 101                         "%s: user is not superuser, "
 102                         "so remain in time-sharing class\n", argv[0]));
 103           else
 104             ACE_ERROR ((LM_ERROR,
 105                         "%s: ACE_OS::sched_params failed\n", argv[0]));
 106         }
 107
 108       if (ACE_OS::thr_setprio (min_priority) == -1)
 109         {
 110           ACE_ERROR ((LM_ERROR, "(%P|%t) main thr_setprio failed,"
 111                       "no real-time features\n"));
 112         }
 113
 114       CORBA::Object_var naming_obj =
 115         orb->resolve_initial_references ("NameService");
 116
 117       if (CORBA::is_nil (naming_obj.in ()))
 118         ACE_ERROR_RETURN ((LM_ERROR,
 119                            " (%P|%t) Unable to get the Naming Service.\n"),
 120                           1);
 121
 122       CosNaming::NamingContext_var naming_context =
 123         CosNaming::NamingContext::_narrow (naming_obj.in ());
 124
 125       CosNaming::Name name (1);
 126       name.length (1);
 127       name[0].id = CORBA::string_dup ("EventService");
 128
 129       CORBA::Object_var ec_obj =
 130         naming_context->resolve (name);
 131
 132       RtecEventChannelAdmin::EventChannel_var channel;
 133       if (CORBA::is_nil (ec_obj.in ()))
 134         channel = RtecEventChannelAdmin::EventChannel::_nil ();
 135       else
 136         channel = RtecEventChannelAdmin::EventChannel::_narrow (ec_obj.in ());
 137
 138       poa_manager->activate ();
 139
 140       this->connect_consumers (channel.in ());
 141
 142       ACE_DEBUG ((LM_DEBUG, "connected consumer(s)\n"));
 143
 144       ACE_DEBUG ((LM_DEBUG, "running the test\n"));
 145       orb->run ();
 146
 147       ACE_DEBUG ((LM_DEBUG, "event loop finished\n"));
 148
 149       this->disconnect_consumers ();
 150
 151       channel->destroy ();
 152     }
 153   catch (const CORBA::SystemException& sys_ex)
 154     {
 155       sys_ex._tao_print_exception ("SYS_EX in Consumer");
 156     }
 157   catch (const CORBA::Exception& ex)
 158     {
 159       ex._tao_print_exception ("NON SYS EX in Consumer");
 160     }
 161   return 0;
 162 }
 163
 164 void
 165 Driver::push_consumer (void* /* consumer_cookie */,
 166                        ACE_hrtime_t /* arrival */,
 167                        const RtecEventComm::EventSet& events)
 168 {
 169   // int ID =
 170   //   (reinterpret_cast<Test_Consumer**> (consumer_cookie)
 171   //    - this->consumers_);
 172   //
 173   // ACE_DEBUG ((LM_DEBUG, "(%t) events received by consumer %d\n", ID));
 174
 175   if (events.length () == 0)
 176     {
 177       // ACE_DEBUG ((LM_DEBUG, "no events\n"));
 178       return;
 179     }
 180
 181   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->recv_count_mutex_);
 182
 183   this->recv_count_ += events.length ();
 184
 185   int x = this->event_count_ / 10;
 186   if (this->recv_count_ % x == 0)
 187     {
 188       ACE_DEBUG ((LM_DEBUG,
 189                   "ECM_Consumer (%P|%t): %d events received\n",
 190                   this->recv_count_));
 191     }
 192
 193   if (this->recv_count_ >= this->event_count_)
 194     {
 195       TAO_ORB_Core_instance ()->orb ()->shutdown ();
 196     }
 197
 198   // ACE_DEBUG ((LM_DEBUG, "%d event(s)\n", events.length ()));
 199
 200 #if (TAO_NO_COPY_OCTET_SEQUENCES == 1)
 201   for (u_int i = 0; i < events.length (); ++i)
 202     {
 203       const RtecEventComm::Event& e = events[i];
 204
 205       if (e.data.payload.mb () == 0)
 206         {
 207           ACE_DEBUG ((LM_DEBUG, "No data in event[%d]\n", i));
 208           continue;
 209         }
 210
 211       // @@ TODO this is a little messy, infortunately we have to
 212       // extract the first byte to determine the byte order, the CDR
 213       // cannot do it for us because in certain cases the byte order
 214       // is not in the encapsulation. Maybe we need another
 215       // constructor for the InputCDR streams (but there are too many
 216       // already!)?
 217
 218       // Note that there is no copying
 219       int byte_order = e.data.payload[0];
 220
 221       ACE_Message_Block* mb =
 222         ACE_Message_Block::duplicate (e.data.payload.mb ());
 223       mb->rd_ptr (1); // skip the byte order
 224
 225       TAO_InputCDR cdr (mb, byte_order);
 226
 227       ECM_IDLData::Info info;
 228       cdr >> info;
 229
 230       ECM_Data other;
 231       cdr >> other;
 232
 233       if (!cdr.good_bit ())
 234         ACE_ERROR ((LM_ERROR, "Problem demarshalling C++ data\n"));
 235
 236       ACE_Message_Block::release (mb);
 237
 238       CORBA::ULong n = info.trajectory.length ();
 239       // ACE_DEBUG ((LM_DEBUG, "Payload contains <%d> elements\n", n));
 240       // ACE_DEBUG ((LM_DEBUG, "Inventory <%s> contains <%d> elements\n",
 241       // other.description.in (),
 242       // other.inventory.current_size ()));
 243
 244       for (CORBA::ULong j = 0; j < n; ++j)
 245         {
 246           ECM_IDLData::Point& p = info.trajectory[j];
 247           if (static_cast<CORBA::ULong>(p.x) != j ||
 248               static_cast<CORBA::ULong>(p.y) != j*j)
 249             {
 250               ACE_DEBUG ((LM_DEBUG,
 251                           "invalid data in trajectory[%d] = (%f,%f)\n",
 252                           j, p.x, p.y));
 253             }
 254         }
 255     }
 256 #endif /* TAO_NO_COPY_OCTET_SEQUENCES == 1 */
 257 }
 258
 259 void
 260 Driver::connect_consumers (RtecEventChannelAdmin::EventChannel_ptr channel)
 261 {
 262   for (int i = 0; i < this->n_consumers_; ++i)
 263     {
 264       char buf[BUFSIZ];
 265       ACE_OS::sprintf (buf, "consumer_%02d", i);
 266
 267       ACE_NEW (this->consumers_[i],
 268                Test_Consumer (this, this->consumers_ + i));
 269
 270       this->consumers_[i]->connect (this->event_a_,
 271                                     this->event_b_,
 272                                     channel);
 273     }
 274 }
 275
 276 void
 277 Driver::disconnect_consumers (void)
 278 {
 279   for (int i = 0; i < this->n_consumers_; ++i)
 280     {
 281       this->consumers_[i]->disconnect ();
 282     }
 283 }
 284
 285 int
 286 Driver::parse_args (int argc, ACE_TCHAR *argv [])
 287 {
 288   ACE_Get_Opt get_opt (argc, argv, ACE_TEXT("dc:n:h:p:"));
 289   int opt;
 290
 291   while ((opt = get_opt ()) != EOF)
 292     {
 293       switch (opt)
 294         {
 295         case 'c':
 296           this->n_consumers_ = ACE_OS::atoi (get_opt.opt_arg ());
 297           break;
 298
 299         case 'n':
 300           this->event_count_ = ACE_OS::atoi (get_opt.opt_arg ());
 301           break;
 302
 303         case 'h':
 304           {
 305             char* aux;
 306                 char* arg = ACE_OS::strtok_r (ACE_TEXT_ALWAYS_CHAR(get_opt.opt_arg ()), ",", &aux);
 307
 308             this->event_a_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
 309                 arg = ACE_OS::strtok_r (0, ",", &aux);
 310             this->event_b_ = ACE_ES_EVENT_UNDEFINED + ACE_OS::atoi (arg);
 311           }
 312           break;
 313
 314         case 'p':
 315           this->pid_file_name_ = get_opt.opt_arg ();
 316           break;
 317
 318         case '?':
 319         default:
 320           ACE_DEBUG ((LM_DEBUG,
 321                       "Usage: %s "
 322                       "[ORB options] "
 323                       "-s <global|local> "
 324                       "-a (send data in events) "
 325                       "-h <args> "
 326                       "-p <pid file name> "
 327                       "\n",
 328                       argv[0]));
 329           return -1;
 330         }
 331     }
 332
 333   if (this->event_count_ <= 0)
 334     {
 335       ACE_DEBUG ((LM_DEBUG,
 336                   "%s: event count (%d) is out of range, "
 337                   "reset to default (%d)\n",
 338                   argv[0], this->event_count_,
 339                   100));
 340       this->event_count_ = 100;
 341     }
 342
 343   if (this->n_consumers_ <= 0)
 344     {
 345       ACE_ERROR_RETURN ((LM_ERROR,
 346                          "%s: number of consumers or "
 347                          "suppliers out of range\n", argv[0]), -1);
 348     }
 349
 350   return 0;
 351 }
 352
 353 // ****************************************************************
 354
 355 Test_Consumer::Test_Consumer (Driver *driver, void *cookie)
 356   : driver_ (driver),
 357     cookie_ (cookie)
 358 {
 359 }
 360
 361 void
 362 Test_Consumer::connect (int event_a,
 363                         int event_b,
 364                         RtecEventChannelAdmin::EventChannel_ptr ec)
 365 {
 366   ACE_ConsumerQOS_Factory qos;
 367   qos.start_disjunction_group ();
 368   qos.insert_type (ACE_ES_EVENT_SHUTDOWN, 0);
 369   qos.insert_type (event_a, 0);
 370   qos.insert_type (event_b, 0);
 371
 372   // = Connect as a consumer.
 373   RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
 374     ec->for_consumers ();
 375
 376   this->supplier_proxy_ =
 377     consumer_admin->obtain_push_supplier ();
 378
 379   RtecEventComm::PushConsumer_var objref = this->_this ();
 380
 381   this->supplier_proxy_->connect_push_consumer (objref.in (),
 382                                                 qos.get_ConsumerQOS ());
 383
 384 }
 385
 386 void
 387 Test_Consumer::disconnect (void)
 388 {
 389   if (CORBA::is_nil (this->supplier_proxy_.in ()))
 390     return;
 391
 392   RtecEventChannelAdmin::ProxyPushSupplier_var proxy =
 393     this->supplier_proxy_._retn ();
 394
 395   proxy->disconnect_push_supplier ();
 396 }
 397
 398 void
 399 Test_Consumer::push (const RtecEventComm::EventSet& events)
 400 {
 401   ACE_hrtime_t arrival = ACE_OS::gethrtime ();
 402   this->driver_->push_consumer (this->cookie_, arrival, events);
 403 }
 404
 405 void
 406 Test_Consumer::disconnect_push_consumer (void)
 407 {
 408 }