Using the Real-Time Event Service

Overview

Contents

1. Overview
2. The COS Event Model
3. Real-Time Event Service Enhancements
4. The Real-Time Event Service
5. Developing Consumers
6. Developing Suppliers
7. Caring For Your Event Channel
8. Sample Applications
9. Reference Materials

Introduction

The standard CORBA operation invocation model supports twoway, oneway, and deferred synchronous interactions between clients and servers. The primary strength of the twoway model is its intuitive mapping onto the object->operation() paradigm supported by OO languages. In principle, twoway invocations simplify the development of distributed applications by supporting an implicit request/response protocol that makes remote operation invocations transparent to the client.

In practice, however, the standard CORBA operation invocation models are too restrictive for real-time applications. In particular, these models lack asynchronous message delivery, do not support timed invocations or group communication, and can lead to excessive polling by clients. Moreover, standard oneway invocations might not implement reliable delivery and deferred synchronous invocations require the use of the CORBA Dynamic Invocation Interface (DII), which yields excessive overhead for most real-time applications.

The Event Service is a CORBA Object Service (COS) that is designed to alleviate some of the restrictions with standard CORBA invocation models. In particular, the COS Event Service supports asynchronous message delivery and allows one or more suppliers to send messages to one or more consumers. Event data can be delivered from suppliers to consumers without requiring these participants to know about each other explicitly.

There are two models (i.e., push vs. pull) of participant collaborations in the COS Event Service architecture. This material focuses on real-time enhancements to the push model, which allows suppliers of events to initiate the transfer of event data to consumers. Suppliers push events to the Event Channel, which in turn pushes the events to consumers.

Suppliers use Event Channels to push data to consumers. Likewise, consumers can explicitly pull data from suppliers. The push and pull semantics of event propagation help to free consumers and suppliers from the overly restrictive synchronous semantics of the standard CORBA twoway communication model. In addition, Event Channels can implement group communication by serving as a replicator, broadcaster, or multicaster that forward events from one or more suppliers to multiple consumers.

• Consumers - Ultimate targets of events
• Suppliers - Generate events
• Event Channel - Decouple consumers and suppliers by propagating events to consumers on behalf of suppliers

Real-Time Event Service Enhancements

• Prioritized dispatching within preemption classes -- The current implementation can dispatch events in the same queue by their order of importance, which is necessary to support priorities within a rate group.

• Suspend/resume -- If a consumer's event dependencies change at run-time, it can utilize the suspend/resume functionality through the new ProxyPushSupplier::suspend and ProxyPushSupplier::resume methods. When a consumer invokes ProxyPushSupplier::suspend, the dependencies registered with that proxy will be disabled until the resume methods is called. These enhancements allow run-time flexibility of event dependencies, but maintains the determinism required by the system scheduling policy (i.e., consumers can not add and remove dependencies at run-time - just suspend and resume them).

• Event data model -- The data model may use unions, untyped buffers, or type anys.

• Event filtering -- Consumers may register for events based on event type, or supplier id. The event channel filters events based on these registrations, to ensure efficient event delivery.

• Event correlation -- Consumers may register for event delivery based on conjunctive or disjunctive sets of events. Conjunctive registrations cause the event channel to notify the consumer when all events in the set have arrived. Disjunctive registrations cause the event channel to notify the consumer when any event in the set has arrived.

• Periodic event processing -- Consumers may register for suppliers based on timed events. Periodic suppliers push events into the channel at well defined intervals.

• Active consumers and suppliers -- See The Real-Time Event Service

  .

The Real-Time Event Service

• Real-time event dispatching
  • Priority-based queueing and preemption mechanisms

• Centralized event filtering and correlation
  • Source/type-based filtering

  • Conjunction/Disjunction (AND/OR) correlations

• Periodic and Aperiodic processing
  • Canonical timeouts

  • Dependency timeouts

Real-Time Event Service Internals

In hard real-time applications, a consumer must also specify RT_Info data for each event it is interested in receiving, and any other events on which that event depends. The RT_Info structure resides in the Scheduler, and is accessed through the scheduler interface. A unique handle is returned to the consumer when an RT_Info is created, which can then be used to set the information in the RT_Info. An RT_Info handle may also be obtained via the Scheduler's lookup method.

• Providing QoS Information
• Connecting Consumers to the Event Channel
• Receiving Events
• Disconnecting Consumers from the Event Channel

  // Obtain a reference to the scheduler server.
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  // Create new RT_Info descriptors for three events.

  RtecScheduler::handle_t handle1 = 
    server->create ("event_1",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );
      
  RtecScheduler::handle_t handle2 = 
    server->create ("event_2",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );

  RtecScheduler::handle_t handle3 = 
    server->create ("event_3",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );


  // Register as a consumer/supplier: act as a supplier of event_1 but with
  // a consumer dependency on event_3.  Therefore, the actual period and 
  // number of threads for event_1 depends on the characteristics of event_3.
  server->set (handle1,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               500,                         // Worst case time (in 100 nanosecs)
               500,                         // Typical time (in 100 nanosecs)
               500,                         // Cached time (in 100 nanosecs)
               0,                           // Period - will depend on event_3
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               0,                           // Threads - will depend on event_3
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);

  // Register as a producer of event_2.
  server->set (handle2,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               500,                         // Worst case time (in 100 nanosecs)
               500,                         // Typical time (in 100 nanosecs)
               500,                         // Cached time (in 100 nanosecs)
               50000 * 10,                  // Period in 100 nsec (= 20 Hz)
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               1,                           // Threads
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);

  // Register as a consumer of event_3.
  server->set (handle3,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               500,                         // Worst case time (in 100 nanosecs)
               500,                         // Typical time (in 100 nanosecs)
               500,                         // Cached time (in 100 nanosecs)
               0,                           // Period - will depend on supplier
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               0,                           // Threads - will depend on supplier
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);


  // Establish a dependency of event_1 on event_3.
  server->add_dependency (handle1,          // handle that depends
                          handle3,          // handle that is depended on
                          1,                // number of calls per event occurance
                          TAO_TRY_ENV       // environment
                         );

void
Test_Consumer::connect (const char* name,
                        int event_a, int event_b,
                        RtecEventChannelAdmin::EventChannel_ptr ec,
                        CORBA::Environment& _env)
{

  // Register operations with the scheduling service.  The following steps are
  // only necessary for applications that make use of the Event Service's hard
  // real-time features.  Applications that do not need these features and are
  // configured with a null scheduler may skip the following operations on the
  // scheduling server.

  // Obtain a reference to the scheduler from the ACE_Scheduler_Factory. 
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  // Create a new RT_Info entry for the function identifier
  // we were passed, and hang onto the handle to the RT_Info.
  RtecScheduler::handle_t rt_info =
    server->create (name, _env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Set the attributes for the RT_Info.  
  ACE_Time_Value tv (0, 2000);
  TimeBase::TimeT time;
  ORBSVCS_Time::Time_Value_to_TimeT (time, tv);
  server->set (rt_info,
               RtecScheduler::VERY_HIGH_CRITICALITY,
               time, time, time,
               0,
               RtecScheduler::VERY_LOW_IMPORTANCE,
               time,
               0,
               RtecScheduler::OPERATION,
               _env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Specify a disjunctive dependency on the arrival of event_a, the arrival
  // of event b, OR the arrival of an event service shutdown event.  Note that 
  // the same RT_Info is used for each event.  This can be used to simplify
  // code in applications using a null scheduler, or to consolidate events
  // with identical characteristics in hard real-time applications.
  ACE_ConsumerQOS_Factory qos;
  qos.start_disjunction_group ();
  qos.insert_type (ACE_ES_EVENT_SHUTDOWN, rt_info);
  qos.insert_type (event_a, rt_info);
  qos.insert_type (event_b, rt_info);

  // = Connect as a consumer.

  // Obtain a reference to the consumer administration object.
  RtecEventChannelAdmin::ConsumerAdmin_var consumer_admin =
    ec->for_consumers (_env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Obtain a reference to the push supplier proxy.
  this->supplier_proxy_ =
    consumer_admin->obtain_push_supplier (_env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Obtain a reference to this object.
  RtecEventComm::PushConsumer_var objref = this->_this (_env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Connect as a consumer.
  this->supplier_proxy_->connect_push_consumer (objref.in (),
                                                qos.get_ConsumerQOS (),
                                                _env);
  TAO_CHECK_ENV_RETURN_VOID(_env);
}

int
ECT_Consumer_Driver::run (int argc, char* argv[])
{
  // argc/argv are used to initialize the ORB and the options
  // for this particular test. Other applications may hard-code
  // the ORB options, obtain them from another source, etc.

  TAO_TRY
    {
      // The use of TAO_TRY macros isolate us from the differences
      // between platforms with and without native C++ exceptions.
      // This is work in progress and may change in the future!

      // Below is some boiler plate code to initialize the ORB and
      // the POA. Notice that applications that connect to the Event
      // Channel play the server role in some instances, because
      // they receive push() requests (as Consumers) or
      // disconnect_push_supplier() requests (as Suppliers).

      // Initialize the ORB reference.
      this->orb_ =
        CORBA::ORB_init (argc, argv, "", TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Initialize the root POA reference.
      CORBA::Object_var poa_object =
        this->orb_->resolve_initial_references("RootPOA");
      if (CORBA::is_nil (poa_object.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to initialize the POA.\n"),
                          1);

      // Obtain the narrowed root POA reference.
      PortableServer::POA_var root_poa =
        PortableServer::POA::_narrow (poa_object.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Obtain a reference to the POA manager.
      PortableServer::POAManager_var poa_manager =
        root_poa->the_POAManager (TAO_TRY_ENV);
      TAO_CHECK_ENV;

       // Now some boiler plate code to obtain a reference to the
       // naming service.....

      // Resolve a reference to the naming service.
      CORBA::Object_var naming_obj =
        this->orb_->resolve_initial_references ("NameService");
      if (CORBA::is_nil (naming_obj.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to get the Naming Service.\n"),
                          1);

      // Narrow the naming service reference.
      CosNaming::NamingContext_var naming_context =
        CosNaming::NamingContext::_narrow (naming_obj.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Use the Naming Service to locate the Scheduling Service and
      // use the Scheduler_Factory to keep a global pointer to the
      // latter. 

      // Initialize the scheduler factory to operate in configuration mode.
      if (ACE_Scheduler_Factory::use_config (naming_context.in ()) == -1)
        return -1;

      // Use the Naming Service to locate the Event Service....

      // Set up the event service lookup name.
      CosNaming::Name name (1);
      name.length (1);
      name[0].id = CORBA::string_dup ("EventService");

      // Resolve a reference to the event service.
      CORBA::Object_var ec_obj =
        naming_context->resolve (name, TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Narrow the reference to the event service.
      RtecEventChannelAdmin::EventChannel_var channel;
      if (CORBA::is_nil (ec_obj.in ()))
        channel = RtecEventChannelAdmin::EventChannel::_nil ();
      else
        channel = RtecEventChannelAdmin::EventChannel::_narrow (ec_obj.in (),
                                                                TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Activate the POA so we can start receiving requests...

      // Activate the POA manager.
      poa_manager->activate (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Connect consumers to the event service.
      this->connect_consumers (channel.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      ACE_DEBUG ((LM_DEBUG, "connected consumer(s)\n"));
      ACE_DEBUG ((LM_DEBUG, "running the test\n"));

      // Run the event loop.
      if (this->orb_->run () == -1)
        ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "orb->run"), -1);
      ACE_DEBUG ((LM_DEBUG, "event loop finished\n"));

      this->dump_results ();

      // Disconnect consumers from the event service.
      this->disconnect_consumers (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Destroy the event service.
      channel->destroy (TAO_TRY_ENV);
      TAO_CHECK_ENV;
    }
  TAO_CATCH (CORBA::SystemException, sys_ex)
    {
      TAO_TRY_ENV.print_exception ("SYS_EX");
    }
  TAO_CATCHANY
    {
      TAO_TRY_ENV.print_exception ("NON SYS EX");
    }
  TAO_ENDTRY;
  return 0;
}

void
Test_Consumer::push (const RtecEventComm::EventSet& events,
                     CORBA::Environment &_env)
{
  // Make sure at least one event was pushed.
  if (events.length () == 0)
    {
      // ACE_DEBUG ((LM_DEBUG, "no events\n"));
      return;
    }

  // Make sure only one thread has access.
  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);

  // We start the timer as soon as we receive the first event.
  if (this->recv_count_ == 0)
    this->timer_.start ();

  // Update the count of received events.
  this->recv_count_ += events.length ();

  if (TAO_debug_level > 0
      && this->recv_count_ % 1000 == 0)
    {
      ACE_DEBUG ((LM_DEBUG,
                  "ECT_Consumer (%P|%t): %d events received\n",
		  this->recv_count_));
    }

  // Loop through the events, looking for shutdown events.
  for (u_int i = 0; i < events.length (); ++i)
    {
      if (events[i].header.type == ACE_ES_EVENT_SHUTDOWN)
        {
          this->shutdown_count_++;
          if (this->shutdown_count_ >= this->n_suppliers_)
            {
              // We stop the timer as soon as we realize it is time to
              // do so.
              this->timer_.stop ();
              this->driver_->shutdown_consumer (this->cookie_, _env);
            }
        }
    }
}

void
Test_Consumer::disconnect (CORBA::Environment &_env)
{
  // Make sure the supplier proxy reference is valid.
  if (CORBA::is_nil (this->supplier_proxy_.in ()))
    return;

  // Disconnect from further communication with the push
  // supplier(s).  Each consumer is represented by a unique
  // ACE_ES_ConsumerModule instance. Which connection to
  // disconnect is determined by the instance for the consumer.
  this->supplier_proxy_->disconnect_push_supplier (_env);
  TAO_CHECK_ENV_RETURN_VOID(_env);

  // Mark the supplier proxy reference invalid.
  this->supplier_proxy_ =
    RtecEventChannelAdmin::ProxyPushSupplier::_nil ();

  // We want to stop processing events for this consumer. Above,
  // we disconnected the consumer from the Event Channel, so no
  // more events will be sent, but we could have some events in
  // transit.

  // Without a flushing protocol we need to deactivate the
  // servant to stop accepting push () requests for any 
  // incoming events.

  // Deactivate the servant
  PortableServer::POA_var poa = 
    this->_default_POA (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
  PortableServer::ObjectId_var id =
    poa->servant_to_id (this, _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
  poa->deactivate_object (id.in (), _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
}

Developing Suppliers

• Providing QoS Information
• Connecting Suppliers to the Event Channel
• Generating Events
• Disconnecting Suppliers from the Event Channel

  // Obtain a reference to the scheduler server.
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  // Create new RT_Info descriptors for two events.

  RtecScheduler::handle_t handle0 = 
    server->create ("event_0",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );

  RtecScheduler::handle_t handle1 = 
    server->create ("event_1",              // Name of entry point
                    TAO_TRY_ENV             // Environment
                   );

  // Register as a producer of event_0.
  server->set (handle0,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               10,                          // Worst case time (in 100 nanosecs)
               10,                          // Typical time (in 100 nanosecs)
               10,                          // Cached time (in 100 nanosecs)
               50000 * 10,                  // Period in 100 nanosecs (= 20 Hz)
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               1,                           // Threads
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);

  // Register as a producer of event_1.
  server->set (handle1,                     // RT_Info handle
               RtecScheduler::HIGH_CRITICALITY,   // Criticality
               10,                          // Worst case time (in 100 nanosecs)
               10,                          // Typical time (in 100 nanosecs)
               10,                          // Cached time (in 100 nanosecs)
               50000 * 10,                  // Period in 100 nanosecs (= 20 Hz)
               RtecScheduler::LOW_IMPORTANCE,     // Importance
               0,                           // Quantum (unused)
               1,                           // Threads
               RtecScheduler::OPERATION,          // Info type
               TAO_TRY_ENV);

Connecting Suppliers to Event Channel

void
Test_Supplier::connect (const char* name,
                        int burst_count,
                        int burst_size,
                        int event_size,
                        int burst_pause,
                        int event_a,
                        int event_b,
                        RtecEventChannelAdmin::EventChannel_ptr ec,
                        CORBA::Environment &_env)
{
  // Some application-specific setup code.

  // Store the passed parameters in the object.
  this->burst_count_ = burst_count;
  this->burst_size_ = burst_size;
  this->event_size_ = event_size;
  this->burst_pause_ = burst_pause;
  this->event_a_ = event_a;
  this->event_b_ = event_b;
   
  // Register operations with the scheduling service.  The following steps are
  // only necessary for applications that make use of the Event Service's hard
  // real-time features.  Applications that do not need these features and are
  // configured with a null scheduler may skip the following operations on the
  // scheduling server.

  // Obtain a reference to the scheduling service. 
  RtecScheduler::Scheduler_ptr server =
    ACE_Scheduler_Factory::server ();

  // Create an RT_Info descriptor for the passed operation name. 
  RtecScheduler::handle_t rt_info =
    server->create (name, _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Calculate the period at which to supply events.
  ACE_Time_Value tv (0, burst_pause);
  RtecScheduler::Period_t rate = tv.usec () * 10;

  // Set the information in the RT_Info descriptor.
  tv.set (0, 2000);
  TimeBase::TimeT time;
  ORBSVCS_Time::Time_Value_to_TimeT (time, tv);
  server->set (rt_info,
               RtecScheduler::VERY_HIGH_CRITICALITY,
               time, time, time,
               rate,
               RtecScheduler::VERY_LOW_IMPORTANCE,
               time,
               1,
               RtecScheduler::OPERATION,
               _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Now, create a supplier id, and publish the events 
  // that will be supplied under this id.

  // Create a supplier id from the passed name
  this->supplier_id_ = ACE::crc32 (name);
  ACE_DEBUG ((LM_DEBUG, "ID for <%s> is %04.4x\n", name,
              this->supplier_id_));

  // Publish the events the supplier provides.
  ACE_SupplierQOS_Factory qos;
  qos.insert (this->supplier_id_,
              event_a,
              rt_info, 1);
  qos.insert (this->supplier_id_,
              event_b,
              rt_info, 1);
  qos.insert (this->supplier_id_,
              ACE_ES_EVENT_SHUTDOWN,
              rt_info, 1);

  // And finally, some boiler plate code to connect a supplier 
  // to the Event Service.  This is where the connection is
  // actually made.

  // Obtain a reference to the supplier administration object.
  RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
    ec->for_suppliers (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Obtain a reference to the consumer proxy object.
  this->consumer_proxy_ =
    supplier_admin->obtain_push_consumer (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Obtain a reference to this supplier object.
  RtecEventComm::PushSupplier_var objref =
    this->supplier_._this (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Connect as a supplier of the published events.
  this->consumer_proxy_->connect_push_supplier (objref.in (),
                                                qos.get_SupplierQOS (),
                                                _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
}

int
ECT_Supplier_Driver::run (int argc, char* argv[])
{
  // argc/argv are used to initialize the ORB and the options
  // for this particular test. Other applications may hard-code
  // the ORB options, obtain them from another source, etc.

  TAO_TRY
    {
      // The use of TAO_TRY macros isolate us from the differences
      // between platforms with and without native C++ exceptions.
      // This is work in progress and may change in the future!

      // Below is some boiler plate code to initialize the ORB and
      // the POA. Notice that applications that connect to the Event
      // Channel play the server role in some instances, because
      // they receive push() requests (as Consumers) or
      // disconnect_push_supplier() requests (as Suppliers).

      // Initialize the ORB reference.
      CORBA::ORB_var orb =
        CORBA::ORB_init (argc, argv, "", TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Initialize the root POA reference.
      CORBA::Object_var poa_object =
        orb->resolve_initial_references("RootPOA");
      if (CORBA::is_nil (poa_object.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to initialize the POA.\n"),
                          1);

      // Obtain the narrowed root POA reference.
      PortableServer::POA_var root_poa =
        PortableServer::POA::_narrow (poa_object.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Obtain a reference to the POA manager.
      PortableServer::POAManager_var poa_manager =
        root_poa->the_POAManager (TAO_TRY_ENV);
      TAO_CHECK_ENV;


       // Now some boiler plate code to obtain a reference to the
       // naming service.....

      // Resolve a reference to the naming service.
      CORBA::Object_var naming_obj =
        orb->resolve_initial_references ("NameService");
      if (CORBA::is_nil (naming_obj.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to get the Naming Service.\n"),
                          1);

      // Narrow the naming service reference.
      CosNaming::NamingContext_var naming_context =
        CosNaming::NamingContext::_narrow (naming_obj.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Use the Naming Service to locate the Scheduling Service and
      // use the Scheduler_Factory to keep a global pointer to the
      // latter. 

      // Initialize the scheduler factory to operate in configuration mode.
      if (ACE_Scheduler_Factory::use_config (naming_context.in ()) == -1)
        return -1;

      // Use the Naming Service to locate the Event Service....

      // Set up the event service lookup name.
      CosNaming::Name name (1);
      name.length (1);
      name[0].id = CORBA::string_dup ("EventService");

      // Resolve a reference to the event service.
      CORBA::Object_var ec_obj =
        naming_context->resolve (name, TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Narrow the reference to the event service.
      RtecEventChannelAdmin::EventChannel_var channel;
      if (CORBA::is_nil (ec_obj.in ()))
        channel = RtecEventChannelAdmin::EventChannel::_nil ();
      else
        channel = RtecEventChannelAdmin::EventChannel::_narrow (ec_obj.in (),
                                                                TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Activate the POA so we can start receiving requests...

      // Activate the POA manager.
      poa_manager->activate (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Connect suppliers to the event service.
      this->connect_suppliers (channel.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      ACE_DEBUG ((LM_DEBUG, "connected supplier(s)\n"));

      // Activate the supplier objects
      this->activate_suppliers (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      ACE_DEBUG ((LM_DEBUG, "suppliers are active\n"));

      // Wait for the supplier threads.
      if (ACE_Thread_Manager::instance ()->wait () == -1)
        {
          ACE_ERROR ((LM_ERROR, "Thread_Manager wait failed\n"));
          return 1;
        }

      ACE_DEBUG ((LM_DEBUG, "suppliers finished\n"));

      this->dump_results ();

      // Disconnect suppliers from the event service.
      this->disconnect_suppliers (TAO_TRY_ENV);
      TAO_CHECK_ENV;
    }
  TAO_CATCH (CORBA::SystemException, sys_ex)
    {
      TAO_TRY_ENV.print_exception ("SYS_EX");
    }
  TAO_CATCHANY
    {
      TAO_TRY_ENV.print_exception ("NON SYS EX");
    }
  TAO_ENDTRY;
  return 0;
}

int
Test_Supplier::svc ()
{
  TAO_TRY
    {
      // First, a bunch of code that is specific to this test.

      // Set pause (sleep) value between message bursts.
      ACE_Time_Value tv (0, this->burst_pause_);

      // Set up message block for event data.
      ACE_Message_Block mb (this->event_size_);
      mb.wr_ptr (this->event_size_);

      // Create an event set for one event, initialize event header.
      RtecEventComm::EventSet event (1);
      event.length (1);
      event[0].header.source = this->supplier_id ();
      event[0].header.ttl = 1;

      // Set up time stamps in event header.  This is for performance
      // measurements, so this step can be omitted at will.
      ACE_hrtime_t t = ACE_OS::gethrtime ();
      ORBSVCS_Time::hrtime_to_TimeT (event[0].header.creation_time, t);
      event[0].header.ec_recv_time = ORBSVCS_Time::zero;
      event[0].header.ec_send_time = ORBSVCS_Time::zero;

      // Initialize data fields in event.
      event[0].data.x = 0;
      event[0].data.y = 0;

      // We use replace to minimize the copies. This should result
      // in just one memory allocation;
      event[0].data.payload.replace (this->event_size_,
                                     &mb);

      // This is where the events are actually pushed into
      // the event channel.  The test pushes bursts of events,
      // pausing a specified interval between bursts.

      // Start the timer, and begin pushing events.
      this->timer_.start ();
      for (int i = 0; i < this->burst_count_; ++i)
        {
          // Send a burst of events.
          for (int j = 0; j < this->burst_size_; ++j)
            {
              if (j % 2 == 0)
                event[0].header.type = this->event_a_;
              else
                event[0].header.type = this->event_b_;

              // ACE_DEBUG ((LM_DEBUG, "(%t) supplier push event\n"));
              this->consumer_proxy ()->push (event, TAO_TRY_ENV);

              TAO_CHECK_ENV;
            }
 
          // Sleep until it's time to send the next burst.
          ACE_OS::sleep (tv);
        }

      // Send a "magic" type of event to inform the consumer that we are 
      // not sending anything else...

      // Send one event shutdown from each supplier
      event[0].header.type = ACE_ES_EVENT_SHUTDOWN;
      this->consumer_proxy ()->push(event, TAO_TRY_ENV);
      TAO_CHECK_ENV;
      this->timer_.stop ();
      
    }
  TAO_CATCH (CORBA::SystemException, sys_ex)
    {
      TAO_TRY_ENV.print_exception ("SYS_EX");
    }
  TAO_CATCHANY
    {
      TAO_TRY_ENV.print_exception ("NON SYS EX");
    }
  TAO_ENDTRY;
  return 0;
}

Disconnecting Suppliers from the Event Channel

void
Test_Supplier::disconnect (CORBA::Environment &_env)
{
  // Make sure the consumer proxy reference is valid.
  if (CORBA::is_nil (this->consumer_proxy_.in ()))
    return;

  // Disconnect communication with the push consumer(s).
  this->consumer_proxy_->disconnect_push_consumer (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);

  // Mark the consumer proxy reference invalid.
  this->consumer_proxy_ =
    RtecEventChannelAdmin::ProxyPushConsumer::_nil ();

  // We need to stop accepting disconnect_push_supplier () requests 
  // for this supplier, before it is safe to destroy the supplier.
  // As required by the CORBA spec, you must explicitly deactivate
  // a servant before destroying it.

  // Deactivate the servant
  PortableServer::POA_var poa = 
    this->supplier_._default_POA (_env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
  PortableServer::ObjectId_var id =
    poa->servant_to_id (&this->supplier_, _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
  poa->deactivate_object (id.in (), _env);
  TAO_CHECK_ENV_RETURN_VOID (_env);
    RtecEventChannelAdmin::ProxyPushConsumer::_nil ();
}

Caring for your Event Channel

int main (int argc, char *argv[])
{
  TAO_TRY
    {
      // argc/argv are used to initialize the ORB and the options
      // for the Event Service executable. Other applications may
      // hard code the ORB options, obtain them from another source, etc.

      // Again the boiler plate code for ORB and POA initialization.

      // Initialize ORB.
      CORBA::ORB_var orb =
        CORBA::ORB_init (argc, argv, "internet", TAO_TRY_ENV);
      TAO_CHECK_ENV;

      if (parse_args (argc, argv) == -1)
        return 1;

      CORBA::Object_var poa_object =
        orb->resolve_initial_references("RootPOA");
      if (CORBA::is_nil (poa_object.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to initialize the POA.\n"),
                          1);

      PortableServer::POA_var root_poa =
        PortableServer::POA::_narrow (poa_object.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      PortableServer::POAManager_var poa_manager =
        root_poa->the_POAManager (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      CORBA::Object_var naming_obj =
        orb->resolve_initial_references ("NameService");
      if (CORBA::is_nil (naming_obj.in ()))
        ACE_ERROR_RETURN ((LM_ERROR,
                           " (%P|%t) Unable to initialize the Naming Service.\n"),
                          1);

      CosNaming::NamingContext_var naming_context =
        CosNaming::NamingContext::_narrow (naming_obj.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Notice the use of auto_ptr<> to automagically manage the
      // destruction of the servant.  When the auto_ptr goes out
      // of scope, its destructor is called, which in turn destroys
      // the servant.

      auto_ptr scheduler_impl;
      RtecScheduler::Scheduler_var scheduler;


      // Create a new servant to implement the Scheduling Service, 
      // register it with the Naming Service, and use the 
      // Scheduler_Factory to keep a global pointer to the new
      // Scheduling Service.

      // This is the name we (potentially) use to register the Scheduling
      // Service in the Naming Service.
      CosNaming::Name schedule_name (1);
      schedule_name.length (1);
      schedule_name[0].id = CORBA::string_dup ("ScheduleService");

      if (global_scheduler == 0)
        {
          scheduler_impl =
            auto_ptr(new ACE_Config_Scheduler);
          if (scheduler_impl.get () == 0)
	    return 1;
          scheduler = scheduler_impl->_this (TAO_TRY_ENV);
          TAO_CHECK_ENV;

	  CORBA::String_var str =
	    orb->object_to_string (scheduler.in (), TAO_TRY_ENV);
	  TAO_CHECK_ENV;
	  ACE_DEBUG ((LM_DEBUG, "The (local) scheduler IOR is <%s>\n",
		      str.in ()));

	  // Register the servant with the Naming Context....
	  naming_context->bind (schedule_name, scheduler.in (), TAO_TRY_ENV);
	  TAO_CHECK_ENV;
        }

      ACE_Scheduler_Factory::use_config (naming_context.in ());

      // The Event Service can be configured to support priority based
      // dispatching (the "default_Module_Factory") or best effort (the
      // "Reactive_Module_Factory"). We pick the right module factory
      // according to the command line options processed above.

      TAO_Default_Module_Factory default_module_factory;
      TAO_Reactive_Module_Factory reactive_module_factory;

      TAO_Module_Factory* module_factory = &default_module_factory;
      if (reactive)
        module_factory = &reactive_module_factory;

      // Now, create a new event channel servant to implement the
      // Event Service, and register it with Naming Service.

      // Construct the event channel using the given module factory.
      ACE_EventChannel ec_impl (1,
                                ACE_DEFAULT_EVENT_CHANNEL_TYPE,
                                module_factory);

      // Obtain an object reference to the new channel.
      RtecEventChannelAdmin::EventChannel_var ec =
        ec_impl._this (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Convert the EC object reference to a string.
      CORBA::String_var str =
        orb->object_to_string (ec.in (), TAO_TRY_ENV);

      // Output the EC object reference string (debug only).
      ACE_DEBUG ((LM_DEBUG,
		  "The EC IOR is <%s>\n", str.in ()));

      // Register the EC with the Naming Service.
      CosNaming::Name channel_name (1);
      channel_name.length (1);
      channel_name[0].id = CORBA::string_dup (service_name);
      naming_context->bind (channel_name, ec.in (), TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Activate the POA so we can start receiving requests...

      // Activate the POA manager.
      poa_manager->activate (TAO_TRY_ENV);
      TAO_CHECK_ENV;

      // Run the ORB event loop
      ACE_DEBUG ((LM_DEBUG, "%s; running event service\n", __FILE__));
      if (orb->run () == -1)
        ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "run"), 1);

      // Now the Event Service is finished.  We could deactivate the 
      // EC and SS here, but we don't need to, as the server is 
      // going down anyway.


      // Remove the event service and the scheduling service from
      // the Naming Service.

      naming_context->unbind (channel_name, TAO_TRY_ENV);
      TAO_CHECK_ENV;

      if (global_scheduler == 0)
	{
	  naming_context->unbind (schedule_name, TAO_TRY_ENV);
	  TAO_CHECK_ENV;
	}

    }
  TAO_CATCHANY
    {
      TAO_TRY_ENV.print_exception ("EC");
    }
  TAO_ENDTRY;


  return 0;
}

Sample Applications

Reference Materials

Books

Papers

• Architectural Considerations for Deterministic Real-Time ORB Endsystems and Applications
• The Design and Performance of a Real-Time CORBA Object Event Service
• The Design of the TAO Real-Time Object Request Broker