5 files changed, 71 insertions, 61 deletions

diff --git a/ciaosvcs/Events/CIAO_EventService_Factory_impl.cpp b/ciaosvcs/Events/CIAO_EventService_Factory_impl.cpp
index b690d5689e7..4b1a3a36e96 100644
--- a/ciaosvcs/Events/CIAO_EventService_Factory_impl.cpp
+++ b/ciaosvcs/Events/CIAO_EventService_Factory_impl.cpp
@@ -56,7 +56,15 @@ namespace CIAO
     }
 
     // Activate the servant
-    CIAO_Event_Service_var service = event_service->_this ();
+    PortableServer::ObjectId_var oid =
+      this->poa_->activate_object (event_service);
+
+    CORBA::Object_var obj = poa_->id_to_reference (oid.in());
+
+    CIAO_Event_Service_var service =
+      CIAO_Event_Service::_narrow (obj.in ());
+
+    //CIAO_Event_Service_var service = event_service->_this ();
     return service._retn ();
   }
 
diff --git a/ciaosvcs/Events/CIAO_EventService_Factory_impl.h b/ciaosvcs/Events/CIAO_EventService_Factory_impl.h
index 349a48aa09a..e973279330d 100644
--- a/ciaosvcs/Events/CIAO_EventService_Factory_impl.h
+++ b/ciaosvcs/Events/CIAO_EventService_Factory_impl.h
@@ -35,6 +35,9 @@ namespace CIAO
     EventService_Factory_impl (CORBA::ORB_ptr orb,
                                PortableServer::POA_ptr poa);
 
+    //void init (CORBA::ORB_ptr orb,
+    //           PortableServer::POA_ptr poa);
+
     virtual ~EventService_Factory_impl (void);
 
     /// A factory method which creates an CIAO_Event_Service object
diff --git a/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.cpp b/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.cpp
index db33f9171b7..4ef814b467c 100644
--- a/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.cpp
+++ b/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.cpp
@@ -24,9 +24,7 @@ namespace CIAO
   RTEventService::RTEventService (CORBA::ORB_ptr orb,
                                   PortableServer::POA_ptr poa) :
     orb_ (CORBA::ORB::_duplicate (orb)),
-    root_poa_ (PortableServer::POA::_duplicate (poa)),
-    type_id_ (ACE_ES_EVENT_ANY),
-    source_id_ (ACE_ES_EVENT_SOURCE_ANY)
+    root_poa_ (PortableServer::POA::_duplicate (poa))
   {
     this->create_rt_event_channel ();
   }
@@ -91,24 +89,13 @@ namespace CIAO
         ACE_THROW (CORBA::BAD_PARAM ());
       }
 
-    ACE_Hash<ACE_CString> hasher;
-    this->source_id_ = hasher (supplier_config->supplier_id (ACE_ENV_SINGLE_ARG_PARAMETER));
-
-    ACE_CHECK;
-    this->type_id_ = this->source_id_;
-
-    if (CIAO::debug_level () > 11)
-      {
-        ACE_DEBUG ((LM_DEBUG, "connect source id: %i\n", this->source_id_));
-      }
-
+    // Get a proxy push consumer from the EventChannel.
     RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
       this->rt_event_channel_->for_suppliers (ACE_ENV_SINGLE_ARG_PARAMETER);
     ACE_CHECK;
 
-    this->proxy_consumer_ =
-      supplier_admin->obtain_push_consumer (ACE_ENV_SINGLE_ARG_PARAMETER);
-    ACE_CHECK;
+    RtecEventChannelAdmin::ProxyPushConsumer_var proxy_push_consumer =
+      supplier_admin->obtain_push_consumer();
 
     // Create and register supplier servant
     RTEventServiceSupplier_impl * supplier_servant = 0;
@@ -122,10 +109,23 @@ namespace CIAO
       rt_config->rt_event_qos (ACE_ENV_SINGLE_ARG_PARAMETER);
     ACE_CHECK;
 
-    this->proxy_consumer_->connect_push_supplier (push_supplier.in (),
-                                                  qos.in ()
-                                                  ACE_ENV_ARG_PARAMETER);
+    ACE_SupplierQOS_Factory supplier_qos;
+    supplier_qos.insert (ACE_ES_EVENT_SOURCE_ANY, ACE_ES_EVENT_ANY, 0, 1);
+
+    supplier_qos.insert (ACE_ES_EVENT_SOURCE_ANY,
+                       ACE_ES_EVENT_ANY,
+                       0,    // handle to the rt_info structure
+                       1);    
+
+    proxy_push_consumer->connect_push_supplier (push_supplier.in (),
+                                                supplier_qos.get_SupplierQOS ()
+                                                ACE_ENV_ARG_PARAMETER);
     ACE_CHECK;
+
+
+    this->proxy_consumer_map_.bind (
+      supplier_config->supplier_id (), 
+      proxy_push_consumer._retn ());
   }
 
   void
@@ -216,16 +216,13 @@ namespace CIAO
   {
     ACE_UNUSED_ARG (connection_id);
 
-    /* @todo
-    if (! CORBA::is_nil (this->proxy_consumer_.in ()))
-      {
-        this->proxy_consumer_->disconnect_push_consumer (
-          ACE_ENV_SINGLE_ARG_PARAMETER);
-        ACE_CHECK;
-      }
+    RtecEventChannelAdmin::ProxyPushConsumer_var proxy_consumer;
+
+    this->proxy_consumer_map_.unbind (connection_id, proxy_consumer);
 
-    // What to do with the consumers?!
-    */
+    proxy_consumer->disconnect_push_consumer (
+      ACE_ENV_SINGLE_ARG_PARAMETER);
+    ACE_CHECK;
   }
 
   void
@@ -256,15 +253,6 @@ namespace CIAO
       {
         ACE_DEBUG ((LM_DEBUG, "------CIAO::RTEventService::push_event------\n"));
       }
-
-    RtecEventComm::EventSet events (1);
-    events.length (1);
-    events[0].header.source = ACE_ES_EVENT_SOURCE_ANY; //this->source_id_;
-    events[0].header.type = ACE_ES_EVENT_ANY; //this->type_id_;
-    events[0].data.any_value <<= ev;
-
-    this->proxy_consumer_->push (events ACE_ENV_ARG_PARAMETER);
-    ACE_CHECK;
   }
 
   void
@@ -294,7 +282,17 @@ namespace CIAO
     ACE_DEBUG ((LM_DEBUG, "******* push event for source string: %s\n", source_id));
     ACE_DEBUG ((LM_DEBUG, "******* push event for source id: %i\n", events[0].header.source));
 
-    this->proxy_consumer_->push (events ACE_ENV_ARG_PARAMETER);
+    RtecEventChannelAdmin::ProxyPushConsumer_var proxy_consumer;
+
+    if (this->proxy_consumer_map_.find (source_id, proxy_consumer) != 0)
+      {
+        ACE_DEBUG ((LM_DEBUG, 
+                    "CIAO (%P|%t) - RTEventService::ciao_push_event, "
+                    "Error in finding the proxy consumer object.\n"));
+        ACE_THROW (Components::BadEventType ());
+      }
+
+    proxy_consumer->push (events ACE_ENV_ARG_PARAMETER);
     ACE_CHECK;
   }
 
@@ -468,7 +466,7 @@ namespace CIAO
       } 
     else 
       {
-        ACE_DEBUG ((LM_DEBUG, "\nUDP Event Handler Port [%d]\n\n", listen_port));
+        ACE_DEBUG ((LM_DEBUG, "\nUDP Event Handler Port [%d]\n", listen_port));
 
         auto_ptr<TAO_ECG_UDP_EH> udp_eh (new TAO_ECG_UDP_EH (receiver.in()));
         udp_eh->reactor (this->orb_->orb_core ()->reactor ());
@@ -486,6 +484,12 @@ namespace CIAO
   }
 
 
+  ::RtecEventChannelAdmin::EventChannel_ptr
+  RTEventService::tao_rt_event_channel (ACE_ENV_SINGLE_ARG_DECL)
+  {
+    return this->rt_event_channel_;
+  }
+
   //////////////////////////////////////////////////////////////////////
   ///                 Supplier Servant Implementation
   //////////////////////////////////////////////////////////////////////
diff --git a/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.h b/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.h
index 7544bf39201..3c217699bb4 100644
--- a/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.h
+++ b/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.h
@@ -130,6 +130,11 @@ namespace CIAO
       ACE_THROW_SPEC ((
         ::CORBA::SystemException));
 
+    virtual ::RtecEventChannelAdmin::EventChannel_ptr tao_rt_event_channel (
+        ACE_ENV_SINGLE_ARG_DECL_WITH_DEFAULTS
+      )
+      ACE_THROW_SPEC ((::CORBA::SystemException));
+
   private:
     // @@ (GD) This is the place where use could provide a parameter
     //         which specifies the event channel service configuration file.
@@ -154,27 +159,15 @@ namespace CIAO
     RtecEventChannelAdmin::EventChannel_var rt_event_channel_;
 
     /**
-     * @var RtecEventComm::EventType type_id_
-     *
-     * The type of event.
-     */
-    RtecEventComm::EventType type_id_;
-
-    /**
-     * @var RtecEventComm::EventSourceID source_id_
-     * @@@ Need to change this into a map, since multiple sources
-     * could be connected to the event channel.
-     *
-     * The supplier id.
-     */
-    RtecEventComm::EventSourceID source_id_;
-
-    /**
-     * @var RtecEventChannelAdmin::ProxyPushConsumer_var proxy_consumer_
+     * @var ACE_Hash_Map_Manager<> proxy_supplier_map_
      *
-     * The proxy consumer to which events are pushed.
+     * Mapping of each event sink to a proxy supplier for disconnect purposes.
      */
-    RtecEventChannelAdmin::ProxyPushConsumer_var proxy_consumer_;
+    ACE_Hash_Map_Manager_Ex<ACE_CString,
+                            RtecEventChannelAdmin::ProxyPushConsumer_var,
+                            ACE_Hash<ACE_CString>,
+                            ACE_Equal_To<ACE_CString>,
+                            ACE_Null_Mutex> proxy_consumer_map_;
 
     /**
      * @var ACE_Hash_Map_Manager<> proxy_supplier_map_
@@ -182,7 +175,7 @@ namespace CIAO
      * Mapping of each event sink to a proxy supplier for disconnect purposes.
      */
     ACE_Hash_Map_Manager_Ex<ACE_CString,
-                            RtecEventChannelAdmin::ProxyPushSupplier_ptr,
+                            RtecEventChannelAdmin::ProxyPushSupplier_var,
                             ACE_Hash<ACE_CString>,
                             ACE_Equal_To<ACE_CString>,
                             ACE_Null_Mutex> proxy_supplier_map_;
diff --git a/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.idl b/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.idl
index 8cf3fcb5947..072bf1f89a1 100644
--- a/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.idl
+++ b/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.idl
@@ -47,6 +47,8 @@ module CIAO
     boolean create_receiver (in string addr_serv_id,
                              in boolean is_multicast,
                              in unsigned short listen_port);
+
+    RtecEventChannelAdmin::EventChannel tao_rt_event_channel ();
   };
 
 };