TAO/examples/Simulator/Event_Supplier/DOVE_Supplier.cpp

   1
   2 //=============================================================================
   3 /**
   4  *  @file    DOVE_Supplier.cpp
   5  *
   6  *  A wrapper around the event service initialization and
   7  *  marshalling
   8  *
   9  *  @author Michael Kircher (mk1@cs.wustl.edu)
  10  */
  11 //=============================================================================
  12
  13
  14 #include "DOVE_Supplier.h"
  15 #include "tao/ORB_Core.h"
  16
  17 // Static pointer member initialization for Singleton.
  18
  19 ACE_Scheduler_Factory::POD_RT_Info *
  20 DOVE_Supplier::pod_rt_info_instance_ = 0;
  21
  22 // Constructor.
  23
  24 DOVE_Supplier::DOVE_Supplier ()
  25   : initialized_ (0),
  26     connected_ (0),
  27     connection_params_list_ (0),
  28     current_connection_params_ (0),
  29     connection_count_ (0),
  30     current_connection_index_ (0),
  31     internal_DOVE_Supplier_ptr_ (0),
  32     MIB_name_ (0)
  33 {
  34   ACE_NEW (internal_DOVE_Supplier_ptr_,
  35            Internal_DOVE_Supplier (this));
  36
  37   if (internal_DOVE_Supplier_ptr_ == 0)
  38     {
  39       ACE_ERROR ((LM_ERROR,
  40                   "DOVE_Supplier::DOVE_Supplier internal "
  41                   "supplier not allocated."));
  42     }
  43 }
  44
  45 // Destructor.
  46
  47 DOVE_Supplier::~DOVE_Supplier ()
  48 {
  49   for (int i = 0; i < this->connection_count_; ++i)
  50     {
  51       delete (this->connection_params_list_ [i]);
  52     }
  53
  54   delete [] this->connection_params_list_;
  55
  56   delete internal_DOVE_Supplier_ptr_;
  57
  58 }
  59
  60 // Initialize the ORB and the connection to the Name Service
  61
  62 int
  63 DOVE_Supplier::init (void)
  64 {
  65   try
  66   {
  67     // Connect to the RootPOA.
  68     CORBA::Object_var poaObject_var =
  69       TAO_ORB_Core_instance()->orb()->resolve_initial_references("RootPOA");
  70
  71     if (CORBA::is_nil (poaObject_var.in ()))
  72       ACE_ERROR_RETURN ((LM_ERROR,
  73                          " (%P|%t) Unable to initialize the POA.\n"),
  74                         -1);
  75
  76     this->root_POA_var_ =
  77       PortableServer::POA::_narrow (poaObject_var.in ());
  78
  79     this->poa_manager_ =
  80        root_POA_var_->the_POAManager ();
  81
  82     // Get the Naming Service object reference.
  83     CORBA::Object_var namingObj_var =
  84       TAO_ORB_Core_instance()->orb()->resolve_initial_references (
  85           "NameService");
  86
  87     if (CORBA::is_nil (namingObj_var.in ()))
  88       ACE_ERROR_RETURN ((LM_ERROR,
  89                         " (%P|%t) Unable to get the Naming Service.\n"),
  90                         -1);
  91
  92     this->namingContext_var_ =
  93       CosNaming::NamingContext::_narrow (namingObj_var.in ());
  94   }
  95   catch (const CORBA::Exception& ex)
  96   {
  97     ex._tao_print_exception ("DOVE_Supplier::init");
  98     return -1;
  99   }
 100
 101   initialized_ = 1;
 102   return 0;
 103 }
 104
 105 int
 106 DOVE_Supplier::connect (const char* MIB_name,
 107                         const char* es_name,
 108                         const char * ss_name,
 109                         ACE_Scheduler_Factory::POD_RT_Info * pod_rt_info)
 110 {
 111   // Initialize the supplier if this has not already been done.
 112   if ((initialized_ == 0) &&  (this->init () == -1))
 113     {
 114       ACE_ERROR_RETURN ((LM_ERROR,
 115                          " (%P|%t) Unable to initialize the DOVE_Supplier.\n"),
 116                         -1);
 117     }
 118
 119
 120   // Grab the default RT_Info settings if others were not provided.
 121   if (pod_rt_info == 0)
 122     {
 123       // Get the default singleton if we were not passed the data
 124       pod_rt_info = DOVE_Supplier::pod_rt_info_instance ();
 125       if (pod_rt_info == 0)
 126         {
 127           ACE_ERROR_RETURN ((LM_ERROR,
 128                              " (%P|%t) Unable to obtain"
 129                              " the default RT_Info data.\n"),
 130                             -1);
 131         }
 132     }
 133
 134   // Save the passed MIB name
 135   MIB_name_ = (MIB_name == 0) ? "MIB_unknown" : MIB_name;
 136
 137   // Create a new connection parameters structure.
 138   Connection_Params * cp_temp = 0;
 139   ACE_NEW_RETURN (cp_temp, Connection_Params, -1);
 140
 141   // Populate the known fields of the new connection params struct.
 142   cp_temp->pod_rt_info_ = *pod_rt_info;
 143   cp_temp->es_name_ = (es_name == 0) ?  "EventService" : es_name;
 144   cp_temp->ss_name_ = (ss_name == 0) ?  "ScheduleService" : ss_name;
 145
 146   // Allocate a new connection parameters pointer array.
 147   // Cannot use ACE_NEW_RETURN here, as we need to clean up
 148   // cp_temp if we fail here, and we need what cp_temp points
 149   // to after the current scope if we succeed here.
 150   Connection_Params ** cp_list_temp;
 151   cp_list_temp =
 152     new Connection_Params * [this->connection_count_ + 1];
 153   if (cp_list_temp == 0)
 154     {
 155       // Avoid a memory leak if we failed to allocate.
 156       delete cp_temp;
 157
 158       ACE_ERROR_RETURN ((LM_ERROR,
 159                          " (%P|%t) DOVE_Supplier::connect could not "
 160                          "reallocate connection params list"),
 161                         -1);
 162     }
 163
 164   // Copy the connection struct pointers from
 165   // the old list (if any) to the new one.
 166   for (int i = 0; i < this->connection_count_; ++i)
 167     {
 168       cp_list_temp [i] =
 169         this->connection_params_list_ [i];
 170     }
 171
 172   // Put a pointer to the new connection params structure
 173   // in the new list, increment the connection params count,
 174   // and point to the latest connection parameters.
 175   cp_list_temp [this->connection_count_] = cp_temp;
 176   this->current_connection_params_ = cp_temp;
 177   current_connection_index_ = connection_count_;
 178   ++ (this->connection_count_);
 179
 180   // Replace the old list of pointers with the new one
 181   delete [] this->connection_params_list_;
 182   this->connection_params_list_ = cp_list_temp;
 183
 184   // Resolve the event service reference.
 185   if (this->get_EventChannel () == -1)
 186     {
 187       ACE_ERROR_RETURN ((LM_ERROR,
 188                          " (%P|%t) Unable to resolve the event service.\n"),
 189                         -1);
 190     }
 191
 192   // Resolve the scheduling service reference.
 193   if (this->get_Scheduler () == -1)
 194     {
 195       ACE_ERROR_RETURN ((LM_ERROR,
 196                          " (%P|%t) Unable to resolve the scheduler.\n"),
 197                         -1);
 198     }
 199
 200   // Connect to the event service as a supplier.
 201   if (this->connect_Supplier () == -1)
 202     {
 203       ACE_ERROR_RETURN ((LM_ERROR,
 204                          " (%P|%t) Unable to connect to the event service.\n"),
 205                         -1);
 206     }
 207
 208   return 0;
 209
 210 }
 211
 212
 213 // This method is invoked after all connect calls are done.
 214
 215 void
 216 DOVE_Supplier::connected ()
 217 {
 218   if (! connected_)
 219     {
 220       // Code to do post-connection-establishment
 221       // one-time logic goes here.
 222
 223       connected_ = 1;
 224     }
 225 }
 226
 227
 228 void
 229 DOVE_Supplier::disconnect ()
 230 {
 231 }
 232
 233
 234 void
 235 DOVE_Supplier::notify (CORBA::Any &message)
 236 {
 237   // Finalize connection establishment no later than the first event notification
 238   if (! connected_)
 239     {
 240       this->connected ();
 241     }
 242
 243   try
 244   {
 245     RtecEventComm::Event event;
 246     event.header.source = SOURCE_ID;
 247     event.header.type = ACE_ES_EVENT_NOTIFICATION;
 248     event.header.ttl = 1;
 249     ACE_hrtime_t creation_time = ACE_OS::gethrtime ();
 250     ORBSVCS_Time::hrtime_to_TimeT (event.header.creation_time, creation_time);
 251     event.header.ec_recv_time = ORBSVCS_Time::zero ();
 252     event.header.ec_send_time = ORBSVCS_Time::zero ();
 253     event.data.any_value = message;
 254
 255     RtecEventComm::EventSet events;
 256     events.length (1);
 257     events[0] = event;
 258
 259     // Now we invoke a RPC
 260     this->current_connection_params_->proxyPushConsumer_var_->push (events);
 261   }
 262   catch (const CORBA::Exception&)
 263   {
 264     ACE_ERROR ((LM_ERROR,
 265                 "DOVE_Supplier::notify: "
 266                 "unexpected exception.\n"));
 267   }
 268 }
 269
 270
 271 // Use the next connection in the list of established connections.
 272
 273 void
 274 DOVE_Supplier::use_next_connection ()
 275 {
 276   if (connection_count_ > 0)
 277     {
 278       current_connection_index_ =
 279         (current_connection_index_ == connection_count_ - 1)
 280         ? 0 : current_connection_index_ + 1;
 281
 282       current_connection_params_ =
 283         connection_params_list_ [current_connection_index_];
 284     }
 285 }
 286
 287
 288 // Use the previous connection in the list of established connections.
 289
 290 void
 291 DOVE_Supplier::use_prev_connection ()
 292 {
 293   if (connection_count_ > 0)
 294     {
 295       current_connection_index_ =
 296         (current_connection_index_ == 0)
 297         ? connection_count_ - 1
 298         : current_connection_index_ - 1;
 299
 300       current_connection_params_ =
 301         connection_params_list_ [current_connection_index_];
 302     }
 303 }
 304
 305
 306
 307 // -------------------- Internal Demo Supplier -----------------------------
 308
 309 DOVE_Supplier::Internal_DOVE_Supplier::Internal_DOVE_Supplier (DOVE_Supplier *impl_ptr)
 310   : impl_ptr_ (impl_ptr)
 311 {
 312 }
 313
 314 // ----------------------------------------------------------------------------
 315
 316 int
 317 DOVE_Supplier::get_Scheduler ()
 318 {
 319   try
 320     {
 321       CosNaming::Name schedule_name (1);
 322       schedule_name.length (1);
 323       schedule_name[0].id =
 324         CORBA::string_dup (this->current_connection_params_->ss_name_);
 325
 326       CORBA::Object_var objref =
 327           namingContext_var_->resolve (schedule_name);
 328
 329       this->current_connection_params_->scheduler_var_ =
 330         RtecScheduler::Scheduler::_narrow(objref.in ());
 331     }
 332   catch (const CORBA::Exception&)
 333     {
 334       current_connection_params_->scheduler_var_ = 0;
 335       ACE_ERROR_RETURN ((LM_ERROR,
 336                          "DOVE_Supplier::get_Scheduler: "
 337                          "error while resolving scheduler %s\n",
 338                          this->current_connection_params_->ss_name_),
 339                         -1);
 340     }
 341
 342   return 0;
 343 }
 344
 345
 346 int
 347 DOVE_Supplier::get_EventChannel ()
 348 {
 349   try
 350   {
 351     // Get a reference to the Event Service
 352     CosNaming::Name channel_name (1);
 353     channel_name.length (1);
 354     channel_name[0].id =
 355       CORBA::string_dup (this->current_connection_params_->es_name_);
 356
 357     CORBA::Object_var eventServiceObj_var =
 358       this->namingContext_var_->resolve (channel_name);
 359
 360     this->current_connection_params_->eventChannel_var_ =
 361        RtecEventChannelAdmin::EventChannel::_narrow (eventServiceObj_var.in());
 362
 363     if (CORBA::is_nil (this->current_connection_params_->eventChannel_var_.in()))
 364       ACE_ERROR_RETURN ((LM_ERROR,
 365                          "The reference to the event channel is nil!"),
 366                          1);
 367   }
 368   catch (const CORBA::Exception& ex)
 369   {
 370     ex._tao_print_exception ("DOVE_Supplier::get_EventChannel");
 371     return -1;
 372   }
 373
 374   return 0;
 375 }
 376
 377
 378 int
 379 DOVE_Supplier::connect_Supplier ()
 380 {
 381   try
 382   {
 383     // Generate the Real-time information descriptor.
 384     this->current_connection_params_->rt_info_ =
 385       this->current_connection_params_->
 386         scheduler_var_->
 387           create (this->current_connection_params_->pod_rt_info_.entry_point);
 388
 389
 390     this->current_connection_params_->scheduler_var_->
 391       set (this->current_connection_params_->rt_info_,
 392            static_cast<RtecScheduler::Criticality_t> (this->current_connection_params_->pod_rt_info_.criticality),
 393            this->current_connection_params_->pod_rt_info_.worst_case_execution_time,
 394            this->current_connection_params_->pod_rt_info_.typical_execution_time,
 395            this->current_connection_params_->pod_rt_info_.cached_execution_time,
 396            this->current_connection_params_->pod_rt_info_.period,
 397            static_cast<RtecScheduler::Importance_t> (this->current_connection_params_->pod_rt_info_.importance),
 398            this->current_connection_params_->pod_rt_info_.quantum,
 399            this->current_connection_params_->pod_rt_info_.threads,
 400            static_cast<RtecScheduler::Info_Type_t> (this->current_connection_params_->pod_rt_info_.info_type));
 401
 402
 403
 404     // Set the publications to report them to the event channel.
 405
 406     CORBA::Short x = 0;
 407     RtecEventChannelAdmin::SupplierQOS qos;
 408     qos.publications.length (1);
 409     qos.publications[0].event.header.source = SOURCE_ID;
 410     qos.publications[0].event.header.type = ACE_ES_EVENT_NOTIFICATION;
 411     qos.publications[0].event.header.ttl = 1;
 412     qos.publications[0].event.header.creation_time = ORBSVCS_Time::zero ();
 413     qos.publications[0].event.header.ec_recv_time = ORBSVCS_Time::zero ();
 414     qos.publications[0].event.header.ec_send_time = ORBSVCS_Time::zero ();
 415     qos.publications[0].event.data.any_value <<= x;
 416     qos.publications[0].dependency_info.number_of_calls = 1;
 417     qos.publications[0].dependency_info.rt_info =
 418       this->current_connection_params_->rt_info_;
 419
 420     // = Connect as a supplier.
 421     this->current_connection_params_->supplierAdmin_var_ =
 422       this->current_connection_params_->eventChannel_var_->for_suppliers ();
 423
 424     this->current_connection_params_->proxyPushConsumer_var_ =
 425       this->current_connection_params_->supplierAdmin_var_->obtain_push_consumer ();
 426
 427     // In calling _this we get back an object reference and register
 428     // the servant with the POA.
 429     RtecEventComm::PushSupplier_var pushSupplier_var =
 430       this->internal_DOVE_Supplier_ptr_->_this ();
 431
 432     // Connect the supplier to the proxy consumer.
 433     ACE_SupplierQOS_Factory::debug (qos);
 434     this->current_connection_params_->
 435       proxyPushConsumer_var_->connect_push_supplier (pushSupplier_var.in (),
 436                                                      qos);
 437   }
 438   catch (const CORBA::Exception& ex)
 439   {
 440     ex._tao_print_exception ("DOVE_Supplier::connect_supplier");
 441     return -1;
 442   }
 443
 444   return 0;
 445
 446 }
 447
 448
 449 // Access the default rt_info singleton.
 450
 451 ACE_Scheduler_Factory::POD_RT_Info *
 452 DOVE_Supplier::pod_rt_info_instance ()
 453 {
 454   if (DOVE_Supplier::pod_rt_info_instance_ == 0)
 455     {
 456       ACE_NEW_RETURN (DOVE_Supplier::pod_rt_info_instance_,
 457                       ACE_Scheduler_Factory::POD_RT_Info,
 458                       0);
 459
 460       // Set up the default data.
 461       DOVE_Supplier::pod_rt_info_instance_->entry_point = "ABC";
 462       DOVE_Supplier::pod_rt_info_instance_->criticality =
 463         RtecScheduler::VERY_LOW_CRITICALITY;
 464       DOVE_Supplier::pod_rt_info_instance_->worst_case_execution_time =
 465         ORBSVCS_Time::zero ();
 466       DOVE_Supplier::pod_rt_info_instance_->typical_execution_time =
 467         ORBSVCS_Time::zero ();
 468       DOVE_Supplier::pod_rt_info_instance_->cached_execution_time =
 469         ORBSVCS_Time::zero ();
 470       DOVE_Supplier::pod_rt_info_instance_->period = 10000000;
 471       DOVE_Supplier::pod_rt_info_instance_->importance =
 472         RtecScheduler::VERY_LOW_IMPORTANCE;
 473       DOVE_Supplier::pod_rt_info_instance_->quantum = ORBSVCS_Time::zero ();
 474       DOVE_Supplier::pod_rt_info_instance_->threads = 1;
 475       DOVE_Supplier::pod_rt_info_instance_->info_type =
 476         RtecScheduler::OPERATION;
 477     }
 478
 479   return DOVE_Supplier::pod_rt_info_instance_;
 480 }