File: events_tutorial.html

package info (click to toggle)
tao 1.0.7-2
links: PTS
area: non-free
in suites: potato
size: 14,028 kB
ctags: 12,427
sloc: cpp: 170,752; sh: 8,916; yacc: 2,056; makefile: 914; lex: 669
file content (1819 lines) | stat: -rw-r--r-- 61,991 bytes
<html>
<head>
<title>Using the Real-Time Event Service</title>
</head>

<BODY text = "#000000"
link="#000fff"
vlink="#ff0f0f"
bgcolor="#ffffff">

<br>
<br>

<center>
<h1>Using the Real-Time Event Service</h1><P>

<A HREF="http://www.cs.wustl.edu/~cdgill/">Chris Gill</a>,
<A HREF="http://www.cs.wustl.edu/~harrison/">Tim Harrison</a>, and 
<A HREF="http://www.cs.wustl.edu/~coryan/">Carlos O'Ryan</a><br><br>

<A HREF="mailto:cdgill@cs.wustl.edu">cdgill@cs.wustl.edu</a>,
<A HREF="mailto:harrison@cs.wustl.edu">harrison@cs.wustl.edu</a>, and
<A HREF="mailto:coryan@cs.wustl.edu">coryan@cs.wustl.edu</a><br>
</center>

<br>
<br>

<P><HR><P>
<!-- ************************************************************ -->

<center>
<table cellpadding=10 border=0 cellspacing=4>
<tr><td>
<h2><a name="Section0">Overview</a></h2>
<BR>
<BR>
<tr><td>
<h3>Contents</h3>
<BR>
<ol>
  <li> <a href="events_tutorial.html#Section0">Overview</a>
  <li> <a href="events_tutorial.html#Section1">The COS Event Model</a>
  <li> <a href="events_tutorial.html#Section2">Real-Time Event Service Enhancements</a>
  <li> <a href="events_tutorial.html#Section3">The Real-Time Event Service</a>
  <li> <a href="events_tutorial.html#Section4">Developing Consumers</a>
  <li> <a href="events_tutorial.html#Section5">Developing Suppliers</a>
  <li> <a href="events_tutorial.html#Section6">Caring For Your Event Channel</a>
  <li> <a href="events_tutorial.html#Section7">Sample Applications</a>
  <li> <a href="events_tutorial.html#Section8">Reference Materials</a>
</ol>

<P><HR><P>
<tr><td>
<h3>Introduction</h3>
<BR>
This material is intended to provide an introduction to the COS Event Model, 
the Real-Time extensions to that model in TAO, and a set of examples that
illustrate the techniques used to build systems using these models. The CORBA
Event Service provides a flexible model for asynchronous communication  among
objects.  However, the standard CORBAEvent Service specification lacks
important features required by real-time applications. These features include 
event filtering, event correlation, and periodic event processing.<P>

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 <code>object->operation()</code> 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. <P>

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 
<A HREF="http://www.cs.wustl.edu/~schmidt/GLOBECOM-96.ps.gz">excessive
overhead</A> for most real-time applications. <P>

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.  <P>

There are two models (<EM>i.e.</EM>, <EM>push</EM> vs. <EM>pull</EM>)
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. <P>

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.<p>

</table>
</center>

<hr><P>
<!-- ************************************************************ -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td align=center>
<center><h2><a name="Section1">The COS Event Model</a></h2></center>
</table></center>

<table cellpadding=1 border=0 cellspacing=4>
<h3>Relationships Between Modules</h3>

The role of each component in the COS Event Model is outlined
below:<P>

<ul>
  <li> <b><i>EventChannel</i></b> -- The EventChannel interface provides 
       two factory methods, which allow applications to obtain consumer and
       supplier administration objects, respectively, and use them to create
       the other objects described below, to connect to the event channel. <P>

  <li> <b><i>SupplierAdmin</i></b> -- The SupplierAdmin interface provides 
       factory methods which create the appropriate supplier proxy objects. <P>

  <li> <b><i>ConsumerAdmin</i></b> -- The ConsumerAdmin interface provides
       factory methods which create the appropriate consumer proxy objects. <P>

  <li> <b><i>ProxyPullSupplier</i></b> -- The ProxyPullSupplier 
       interface is used by consumers in the pull model to connect and
       disconnect from the channel. This interface inherits from the
       PullSupplier interface, and acts as a proxy for the suppliers
       from which the channel will pull events. <P>

  <li> <b><i>ProxyPushSupplier</i></b> -- The ProxyPushSupplier 
       interface is used by consumers in the push model to connect and
       disconnect from the channel. This interface inherits from the
       PushSupplier interface, and acts as a proxy for the suppliers
       which will push events to the channel. <P>

  <li> <b><i>ProxyPullConsumer</i></b> -- The ProxyPullConsumer 
       interface is used by suppliers in the pull model to connect and
       disconnect from the channel. This interface inherits from the
       PullConsumer interface, and acts as a proxy for the consumers
       on whose behalf the channel will pull events. <P>

  <li> <b><i>ProxyPushConsumer</i></b> -- The ProxyPushConsumer 
       interface is used by suppliers in the push model to connect and
       disconnect from the channel. This interface inherits from the
       PushConsumer interface, and acts as a proxy for the consumers
       to which the channel will push events. <P>

  <li> <b><i>PullSupplier</i></b> -- The PullSupplier interface provides
       the necessary methods of a supplier of events in the pull model. <P>

  <li> <b><i>PushSupplier</i></b> -- The PushSupplier interface provides
       the necessary methods of a supplier of events in the push model. <P>

  <li> <b><i>PullConsumer</i></b> -- The PullConsumer interface provides
       the necessary methods of a consumer of events in the pull model. <P>

  <li> <b><i>PushConsumer</i></b> -- The PushConsumer interface provides
       the necessary methods of a consumer of events in the push model. <P>
</ul>

<tr><td>
</table>
<br>
<br>
<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td align=center>
<center><IMG SRC="http://www.cs.wustl.edu/~schmidt/gifs/ec_participants2.gif"></center>
</table></center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<h3>The Push Model</h3>
<tr><td>
<center><IMG SRC="http://www.cs.wustl.edu/~schmidt/gifs/ec_push.gif"></center>

<tr><td>
<UL>
  <LI> Consumers - Ultimate targets of events
  <LI> Suppliers - Generate events
  <LI> Event Channel - Decouple consumers and suppliers by propagating
       events to consumers on behalf of suppliers
</UL>
</table></center>

<P><HR><P>
<!-- ************************************************************ -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>

<h2><a name="Section2">Real-Time Event Service Enhancements</a></h2>

<tr><td>
<ul>
  <li> <b><i>Prioritized dispatching within preemption
       classes</i></b> -- 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. <p>

  <li> <b><i>Suspend/resume</i></b> -- If a consumer's event
       dependencies change at run-time, it can utilize the
       suspend/resume functionality through the new 
       <code>ProxyPushSupplier::suspend</code> and
       <code>ProxyPushSupplier::resume</code> methods.  When a
       consumer invokes <code>ProxyPushSupplier::suspend</code>, the
       dependencies registered with that proxy will be disabled until
       the <code>resume</code> 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).<p>

  <LI> <B><I>Event data model</I></B> -- The data model may use
       unions, untyped buffers, or type anys.<P>

<LI> <B><I>Event filtering</I></B> -- 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.<P>

<LI> <B><I>Event correlation</I></B> -- 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 <I>all</I> 
events in the set have arrived.  Disjunctive registrations cause the event channel
to notify the consumer when <I>any</I> event in the set has arrived.<P>

<LI> <B><I>Periodic event processing</I></B> -- Consumers
may register for suppliers based on timed events.  Periodic suppliers push events
into the channel at well defined intervals.<P>

  <LI> <B><I>Active consumers and suppliers</I></B> -- See 
       <a href="events_tutorial.html#Section3">The Real-Time Event Service</a><P>.

</ul>
</table>
</center>

<P><HR><P>
<!-- ************************************************************ -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td align=center>
<h2><a name="Section3">The Real-Time Event Service</a></h2>
</table></center>

<center><table cellpadding=1 border=0 cellspacing=4>
<h3>Real-Time ORB and Services</h3>
<tr><td>
<center><IMG SRC="http://www.cs.wustl.edu/~schmidt/gifs/corbaopt10.gif"></center>

<br>
<center><tr><td>
<ul>
  <li> Real-time event dispatching<P>
       <UL>
	 <LI> Priority-based queueing and preemption mechanisms<P>
       </UL>
  <li> Centralized event filtering and correlation<P>
       <ul>
	 <LI> Source/type-based filtering<P>
	 <LI> Conjunction/Disjunction (AND/OR) correlations<P>
       </ul>
	      
  <li> Periodic and Aperiodic processing <P>
       <UL>
	 <LI> Canonical timeouts<P>
	 <LI> Dependency timeouts
       </UL>
</ul>
</center>
</table></center>
<br>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td align=center>
<h3>Real-Time Event Service Internals</h3>
<BR>
<tr><td>
<center><IMG SRC="http://www.cs.wustl.edu/~schmidt/gifs/ec_dispatching.gif"></center>
</table></center>
<br>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<h3>Characteristics of Real-Time Push Event Service Participants</h3>

<tr><td>
Two major roles are played by the participants in a real-time push event 
service.<P>  

<tr><td>
The first role is that of an event consumer, which receives events from the 
event channel.  A consumer specifies the type and/or source id for each event
it is interested in receiving.<P>

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.<P>

<tr><td>
The second role is that of an event supplier, which generates events and 
passes them into the event channel.  A supplier must specify its source id,
and the type of each event it will generate.

<tr><td>
In hard real-time applications, a supplier must also specify RT_Info data for
the events it will generate.  In particular, it must specify the maximum rate 
at which it will generate each event.  This information is used by a real-time
scheduler to assign appropriate dispatch priorities.<P>

<tr><td>
Note that the event channel may also be configured to use a null scheduling service. This will
cause all operations to be dispatched at the same priority, and will not require the application
to specify worst case execution times, periods, etc.

<tr><td>
The consumer and supplier roles may be combined, as illustrated in the tables below.  There are
two main cases in which the roles are combined: a passive one termed Consumer/Supplier which
borrows a thread  of execution to produce events, and an active one termed Supplier/Consumer
which produces events in its own thread. Both consume events and produce events.<P>
</table></center>

<br>
<br>
<center><table cellpadding=4 border=1 cellspacing=4>
<h4>EC Roles and Specified RT_Info Contents</h4>

<tr>
<th>EC Roles
<th>RT_Info Contents
<th>Domain Examples 
</tr>

<tr>
<td> Consumer
<td> dependencies (and optionally, importance)
<td> Display, Exception & Maintenance Logs
</tr>

<tr>
<td> Consumer/Supplier
<td> dependencies (and optionally, importance)
<td> Navigation Component  (NAV)
</tr>

<tr>
<td> Supplier/Consumer
<td> rate, dependencies (and optionally, importance)
<td> Kalman Filter
</tr>

<tr>
<td> Supplier
<td> rate
<td> Operator Control Panel, EC Reactor Threads 
</tr>

</table></center>
<BR>
<BR>


<center><table cellpadding=1 border=1 cellspacing=4>
<h4>EC Roles and Scheduler Dependency Chain</h4>

<tr>
<th> EC Roles
<th> Scheduler Dependency Chain
</tr>

<tr>
<td> Pure Consumer
<td> root node
</tr>

<tr>
<td> Consumer/Supplier
<td> internal node
</tr>

<tr>
<td> Supplier/Consumer  
<td> internal node
</tr>

<tr>
<td> Pure Supplier
<td> leaf node
</tr>

</table></center>
<BR>
<BR>

<center><table cellpadding=1 border=1 cellspacing=4>
<h4>EC Roles, Threading, and CORBA Roles</h4>

<tr>
<th> EC Roles
<th> Activity
<th> Thread Behavior
<th> CORBA Roles
</tr>

<tr>
<td> Pure Consumer
<td> Passive
<td> Threads optional, "internal", wait for an event to occur
<td> Servant
</tr>

<tr>
<td> Consumer/Supplier
<td> Passive
<td> Threads optional, "internal", wait for an event to occur
<td> Client and/or Servant
</tr>

<tr>
<td> Supplier/Consumer
<td> Active
<td> Threads required and visible to EC: consume events and actively produce other events
<td> Client and/or Servant
</tr>

<tr>
<td> Pure Supplier
<td> Active
<td> Threads required and visible to EC: actively produce events
<td> Client
</tr>
</table></center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>
<center><IMG SRC="http://www.cs.wustl.edu/~cdgill/ec_roles.GIF"></center>
</table></center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>
<center><IMG SRC="http://www.cs.wustl.edu/~cdgill/ec_roles2.GIF"></center>
</table></center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>
<center><IMG SRC="http://www.cs.wustl.edu/~cdgill/ec_backbone.GIF"></center>
</table></center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>
<center><IMG SRC="http://www.cs.wustl.edu/~cdgill/distributed_ec_roles.GIF"></center>
</table></center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>
<center><IMG SRC="http://www.cs.wustl.edu/~cdgill/ec_orb_view.GIF"></center>
</table></center>

<P><HR><P>
<!-- ************************************************************ -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>
<center><h2><a name="Section4">Developing Consumers</a></h2></center>
<br>
<ul>
  <li> <a href="events_tutorial.html#Section4_1">Providing QoS Information</a>
  <li> <a href="events_tutorial.html#Section4_2">Connecting Consumers to the Event Channel</a>
  <li> <a href="events_tutorial.html#Section4_3">Receiving Events</a>
  <li> <a href="events_tutorial.html#Section4_4">Disconnecting Consumers from the Event Channel</a>
</ul>
</table></center>

<P><hr width="75%"><P> <!-- intra-section separator -->
<center><table cellpadding=1 border=0 cellspacing=4>
<center><h3><a name="Section4_1">Providing QoS Information</a><h3></center><p>

<tr><td>
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.

<tr><td>
For each operation, a Consumer should provide the worst case, expected, and 
cached execution time for that operation.  It must also specify criticality and 
importance values for each operation.  A real-time scheduler uses this information 
to order dispatches within a set of operations whose dependencies have been met.

<tr><td>
If it is a Consumer/Supplier (one which consumes an event and passively generates 
one or more events from the thread in which it was called, as illustrated in the
tables above), it must provide dependencies on one or more other events to the
scheduler.  

<tr><td>
If it is a Supplier/Consumer (one which consumes an event and actively generates 
one or more events from its own thread, as illustrated in the tables above), it 
must also specify the rate at which it will generate the new events by passing a
positive value in the period argument to the scheduler set method.  It may also
indicate a positive number of threads in which the dispatch will be made.  If the
number of threads given is zero, but a period is specified, the number of threads
defaults to 1.

</table>
</center>


<center><table cellpadding=1 border=0 cellspacing=4>

<tr><td>
<pre>

  // 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
                         );


</pre>
</table>
</center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<h3>
<a name="Section4_2">Connecting Consumers to the Event Channel</a>
</h3>

<tr><td>
The following code is derived from the EC_Throughput consumer code, which can be 
found in TAO in the file: 
<A HREF="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp">
$TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp</a>

<tr><td>
<pre>

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);
}

</pre>

<tr><td>
The following code is derived from the EC_Throughput consumer driver code, which
can be found in TAO in the file: 
<A HREF="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.cpp">
$TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Consumer_Driver.cpp</a>

<tr><td>
<pre>

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;
}

</pre>
</table>
</center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<!-- Must be after suppliers connect. -->

<center><table cellpadding=1 border=0 cellspacing=4>
<h3><a name="Section4_3">Receiving Events</a></h3><p>

<tr><td>
The following code is derived from the EC_Throughput consumer code, which
can be found in TAO in the file: 
<A HREF="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp">
$TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp</a>

<tr><td>
<pre>

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 (ACE_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);
            }
        }
    }
}

</pre>
</table>
</center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<h3>
<a name="Section4_4">Disconnecting Consumers from the Event Channel</a>
</h3>

<tr><td>
The following code is derived from the EC_Throughput consumer code, which can be 
found in TAO in the file: 
<A HREF="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp">
$TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Consumer.cpp</a>

<tr><td>
<pre>

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);
}

</pre>
</table>
</center>


<P><HR><P>
<!-- ************************************************************ -->

<center><table cellpadding=1 border=0 cellspacing=4>
<td>
<h2><a name="Section5">Developing Suppliers</a></h2>

<ul>
  <li> <a href="events_tutorial.html#Section5_1">Providing QoS Information</a>
  <li> <a href="events_tutorial.html#Section5_2">Connecting Suppliers to the Event Channel</a>
  <li> <a href="events_tutorial.html#Section5_3">Generating Events</a>
  <li> <a href="events_tutorial.html#Section5_4">Disconnecting Suppliers from the Event Channel</a>
</ul>
</table>
</center>


<center><table cellpadding=1 border=0 cellspacing=4>
<h3><a name="Section5_1">Providing QoS Information</a></h3><p>
<tr><td>
In applications that use hard real-time scheduling, a Supplier should provide
the worst case, expected, and  cached execution time for each operation on the
supplier side.  Even if these values are small and highly deterministic, it is
generally better to specify them in the supplier's RT_Info rather than folding
them into the RT_Info of each consumer.

<tr><td>
Such a supplier must also specify criticality and importance values, a period,
and the number of threads for each operation.  A real-time scheduler propagates
this information to consumer RT_Infos along the graph of dependencies.  The
scheduler then uses the propagated information to order dispatches within a
set of operations whose dependencies have been met.

<tr><td>
The Event Service matches supplier publications with consumer subscriptions to
provide efficient event filtering. Providing incorrect publications or
subscriptions will result in missed events.  The Event Service also uses the
subscription information to create additional dependencies between registered 
RT_Infos.  Thus, providing correct supplier publication and consumer
subscription information is also critical for correct scheduling in hard
real-time applications.


<tr><td>
As noted before in the discussion of consumers, 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.

</table>
</center>

<center><table cellpadding=1 border=0 cellspacing=4>

<tr><td>
<pre>

  // 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);

</pre>
</table>
</center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<td>
<h3><a name="Section5_2">Connecting Suppliers to Event Channel</a></h3><P>
<p>

<tr><td>
The following code is derived from the EC_Throughput supplier code, which
can be found in TAO in the file: 
<A HREF="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp">
$TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp</a>


<tr><td>
<pre>

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);
}

</pre>


<tr><td>
The following code is derived from the EC_Throughput supplier driver code, which
can be found in TAO in the file: 
<A HREF="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.cpp">
$TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Supplier_Driver.cpp</a>

<tr><td>
<pre>

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;
}

</pre>
</table>
</center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<h3><a name="Section5_3">Generating Events</a></h3><P><p>

<tr><td>
The following code is derived from the EC_Throughput supplier code, which
can be found in TAO in the file: 
<A HREF="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp">
$TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp</a>

<tr><td>
<pre>

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;
}

</pre>
</table>
</center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<td>
<h3><a name="Section5_4">Disconnecting Suppliers from the Event Channel</a></h3><P>
<p>

<tr><td>
The following code is derived from the EC_Throughput supplier code, which
can be found in TAO in the file: 
<A HREF="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp">
$TAO_ROOT/orbsvcs/tests/EC_Throughput/ECT_Supplier.cpp</a>


<tr><td>
<pre>

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 ();
}

</pre>

</table>
</center>

<P><HR><P>
<!-- ************************************************************ -->

<center><table cellpadding=1 border=0 cellspacing=4>
<td>
<h2><a name="Section6">Caring for your Event Channel</a></h2>


<BR>

<ul>
  <li> <a href="events_tutorial.html#Section6_1">Creation and Name Service Registration</a>
  <li> <a href="events_tutorial.html#Section6_2">Register Suppliers Before Consumers</a>
</ul>
</table>
</center>

<center><table cellpadding=1 border=0 cellspacing=4>
<h3><a name="Section6_1">Creation and Name Service Registration</a><h3><p>

<tr><td>
The following code is derived from the Event_Service executable, which can be
found in TAO in the file: 
<A HREF="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/Event_Service/Event_Service.cpp">
$TAO_ROOT/orbsvcs/Event_Service/Event_Service.cpp</a>

<tr><td>
<pre>

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<POA_RtecScheduler::Scheduler> 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<POA_RtecScheduler::Scheduler>(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;
}

</pre>
</table>
</center>

<P><hr width="75%"><P> <!-- intra-section separator -->

<center><table cellpadding=1 border=0 cellspacing=4>
<h3><a name="Section6_2">Register Suppliers before Consumers</a></h3><p>
<BR>
<BR>
<tr><td>
<center>
<IMG SRC="http://www.cs.wustl.edu/~schmidt/gifs/ec_registration.gif">
</center>
</table></center>

<P><HR><P>
<!-- ************************************************************ -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>
<h2><a name="Section7">Sample Applications</a></h2>

<tr><td> 
A number of sample applications are available in the directories under 
<a href="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/">
TAO's ORB Services tests</a>.

<tr><td>
In particular, much of the code shown in this tutorial was drawn from the
<a href="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Throughput/"> 
EC_Throughput</a> test.  This test exercises the Event Service and measures its
throughput capabilities.

<tr><td>
A similar test,
<a href="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/Event_Latency/"> 
Event_Latency</a>, measures the latency of events through the Event Service.

<tr><td>
The
<a href="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Basic/"> 
EC_Basic</a> test demonstrates the basic use the Event Service.

<tr><td>
The
<a href="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/tests/EC_Multiple/"> 
EC_Multiple</a> test shows a number of ways to connect multiple Event Channels.


<tr><td>
For the IDL source for the various interfaces, please see
<a href="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/orbsvcs/RtecScheduler.idl">RtecScheduler.idl</a>,
<a href="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/orbsvcs/CosEventChannelAdmin.idl">CosEventChannelAdmin.idl</a>,
<a href="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/orbsvcs/CosEventComm.idl">CosEventComm.idl</a>
and
<a href="http://www.cs.wustl.edu/~schmidt/ACE_wrappers/TAO/orbsvcs/orbsvcs/CosNaming.idl">CosNaming.idl</a>.


</table>
</center>

<P><HR><P>
<!-- ************************************************************ -->

<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>
<h2><a name="Section8">Reference Materials</a></h2>

<tr><td> 
The following materials were used in developing this tutorial: please refer to them for further information.

<tr><td> 
<br>
<tr><td> 
<h3>Books</h3>

<tr><td> 
Mowbray, T. and Zahavi, R. The Essential CORBA, Systems Integration Using Distributed Objects.  Wiley, 1995. 
ISBN 0-471-10611-9

<tr><td> 
<br>

<tr><td> 
Baker, S.  CORBA Distributed Objects Using Orbix.  Addison-Wesley, 1997.  ISBN 0-201-92475-7

<tr><td> 
<br>
<tr><td> 
<h3>Papers</h3>

<tr><td>
<ul>

<li><a href="http://www.cs.wustl.edu/~schmidt/Arch.ps.gz">
Architectural Considerations for Deterministic Real-Time ORB
Endsystems and Applications</a>

<li><a href="http://www.cs.wustl.edu/~schmidt/JSAC-98.ps.gz"> The
Design and Performance of a Real-Time CORBA Object Event Service</a>

<li><a href="http://www.cs.wustl.edu/~schmidt/RT-ORB.ps.gz"> The
Design of the TAO Real-Time Object Request Broker</a>

</ul>


<tr><td>

</table>
</center>

<P><HR><P>
<!-- ************************************************************ -->


<center><table cellpadding=1 border=0 cellspacing=4>
<tr><td>
Last modified 10:50:30 CST 22 December 1998 by Chris Gill
</table></center>

</body></html>