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