Fri Apr 21 12:44:42 UTC 2006 Gan Deng <gan.deng@vanderbilt.edu>

4 files changed, 142 insertions, 21 deletions

diff --git a/TAO/CIAO/ChangeLog b/TAO/CIAO/ChangeLog
index 71d77e2f0e5..adadd0922d2 100644
--- a/TAO/CIAO/ChangeLog
+++ b/TAO/CIAO/ChangeLog
@@ -1,3 +1,18 @@
+Fri Apr 21 12:44:42 UTC 2006  Gan Deng  <gan.deng@vanderbilt.edu>
+
+        * ciao/Deployment_Core.idl
+        * ciaosvcs/Events/CIAO_Events_Base/CIAO_Events.idl
+        * ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.cpp
+        * ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.h
+        
+          Modified the CIAO RT Event service implementation by
+          allow Supplier_Config inherits from the EventConsumerBase,
+          so a component servant with a publisher port will cache
+          the object reference of Supplier_Config object. Once a
+          push_event call is made on this object, it will in turn
+          invoke push opeation on the cached proxy_consumer object of
+          the event channel.
+
 Thu Apr 20 01:40:11 UTC 2006  Gan Deng  <gan.deng@vanderbilt.edu>
 
         * ciao/Deployment_Core.idl
diff --git a/TAO/CIAO/ciaosvcs/Events/CIAO_Events_Base/CIAO_Events.idl b/TAO/CIAO/ciaosvcs/Events/CIAO_Events_Base/CIAO_Events.idl
index bcfcca9ef93..3a00cab3183 100644
--- a/TAO/CIAO/ciaosvcs/Events/CIAO_Events_Base/CIAO_Events.idl
+++ b/TAO/CIAO/ciaosvcs/Events/CIAO_Events_Base/CIAO_Events.idl
@@ -25,19 +25,22 @@ module CIAO
     RTNOTIFY
   };
 
-  interface Supplier_Config
+  interface Supplier_Config : Components::EventConsumerBase
   {
     attribute CONNECTION_ID supplier_id;
 
     readonly attribute EventServiceType service_type;
 
     void destroy ();
+
+    void register_proxy_consumer (in Object proxy_consumer);
   };
 
-  interface Consumer_Config : Supplier_Config
+  interface Consumer_Config
   {
     attribute CONNECTION_ID consumer_id;
     attribute Components::EventConsumerBase consumer;
+    readonly attribute EventServiceType service_type;
 
     void start_conjunction_group (in long size);
 
@@ -46,9 +49,11 @@ module CIAO
     void insert_source (in CONNECTION_ID source_id);
 
     void insert_type (in long event_type);
+
+    void destroy ();
   };
 
-  interface CIAO_Event_Service : Components::EventConsumerBase
+  interface CIAO_Event_Service
   {
     Supplier_Config create_supplier_config ();
 
diff --git a/TAO/CIAO/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.cpp b/TAO/CIAO/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.cpp
index e85c8f0ee4f..1ca9ae7d502 100644
--- a/TAO/CIAO/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.cpp
+++ b/TAO/CIAO/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.cpp
@@ -73,8 +73,10 @@ namespace CIAO
     if (CIAO::debug_level () > 9)
       {
         ACE_DEBUG ((LM_DEBUG, "CIAO::RTEventService::connect_event_supplier\n"));
+        ACE_DEBUG ((LM_DEBUG, "Supplier's source id: %s\n",
+                    supplier_config->supplier_id()));
       }
-
+/*
     RTEvent_Supplier_Config_ptr rt_config =
       RTEvent_Supplier_Config::_narrow (supplier_config
                                         ACE_ENV_ARG_PARAMETER);
@@ -84,22 +86,17 @@ 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_));
-      }
-
     RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
       this->rt_event_channel_->for_suppliers (ACE_ENV_SINGLE_ARG_PARAMETER);
     ACE_CHECK;
 
-    this->proxy_consumer_ =
+    RtecEventChannelAdmin::ProxyPushConsumer_var proxy_consumer =
       supplier_admin->obtain_push_consumer (ACE_ENV_SINGLE_ARG_PARAMETER);
     ACE_CHECK;
 
@@ -111,14 +108,25 @@ namespace CIAO
       supplier_servant->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
     ACE_CHECK;
 
-    RtecEventChannelAdmin::SupplierQOS_var qos =
-      rt_config->rt_event_qos (ACE_ENV_SINGLE_ARG_PARAMETER);
+    supplier_config->register_proxy_consumer (proxy_consumer.in ());
+
+    ACE_SupplierQOS_Factory qos;
+
+    qos.insert (this->source_id_,
+                ACE_ES_EVENT_ANY,
+                0,
+                1);
+
+    proxy_consumer->connect_push_supplier (push_supplier.in (),
+                                           qos.get_SupplierQOS ()
+                                           ACE_ENV_ARG_PARAMETER);
     ACE_CHECK;
 
-    this->proxy_consumer_->connect_push_supplier (push_supplier.in (),
-                                                  qos.in ()
-                                                  ACE_ENV_ARG_PARAMETER);
+    ACE_CString supplier_id =
+      supplier_config->supplier_id (ACE_ENV_SINGLE_ARG_PARAMETER);
     ACE_CHECK;
+
+    this->proxy_consumer_map_.bind (supplier_id.c_str (), proxy_consumer._retn ());
   }
 
   void
@@ -170,7 +178,6 @@ namespace CIAO
 
     //@@@
     rt_config->start_disjunction_group (1);
-
     rt_config->insert_type (ACE_ES_EVENT_ANY);
 
     RtecEventChannelAdmin::ConsumerQOS_var qos =
@@ -197,13 +204,13 @@ namespace CIAO
       CORBA::SystemException,
       Components::InvalidConnection))
   {
-    ACE_UNUSED_ARG (connection_id);
+    RtecEventChannelAdmin::ProxyPushConsumer_var proxy_consumer;
+
+    this->proxy_consumer_map_.unbind (connection_id, proxy_consumer);
 
-    this->proxy_consumer_->disconnect_push_consumer (
+    proxy_consumer->disconnect_push_consumer (
       ACE_ENV_SINGLE_ARG_PARAMETER);
     ACE_CHECK;
-
-    // What to do with the consumers?!
   }
 
   void
@@ -241,6 +248,9 @@ namespace CIAO
     events[0].header.type = ACE_ES_EVENT_ANY; //this->type_id_;
     events[0].data.any_value <<= ev;
 
+    // We need to find the right "proxy_consumer" object from the
+    // proxy_consumer_map_, and then call push on it.
+
     this->proxy_consumer_->push (events ACE_ENV_ARG_PARAMETER);
     ACE_CHECK;
   }
@@ -290,6 +300,32 @@ namespace CIAO
     this->_remove_ref ();
   }
 
+/*
+  void
+  RTEventServiceSupplier_impl::push_event (
+      Components::EventBase * ev
+      ACE_ENV_ARG_DECL)
+    ACE_THROW_SPEC ((
+      CORBA::SystemException))
+  {
+    if (CIAO::debug_level () > 10)
+      {
+        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;
+
+    // We need to find the right "proxy_consumer" object from the
+    // proxy_consumer_map_, and then call push on it.
+
+    this->proxy_consumer_->push (events ACE_ENV_ARG_PARAMETER);
+    ACE_CHECK;
+  }
+*/
   //////////////////////////////////////////////////////////////////////
   ///                   Consumer Servant Implementation
   //////////////////////////////////////////////////////////////////////
@@ -426,6 +462,43 @@ namespace CIAO
     this->_remove_ref ();
   }
 
+  void
+  RTEvent_Supplier_Config_impl::register_proxy_consumer (
+      ::CORBA::Object_ptr proxy_consumer
+      ACE_ENV_ARG_DECL_WITH_DEFAULTS
+    )
+    ACE_THROW_SPEC ((::CORBA::SystemException))
+  {
+    // Do a narrow first...
+    this->proxy_consumer_ =
+      RtecEventChannelAdmin::ProxyPushConsumer::_narrow (proxy_consumer);
+  }
+
+  void
+  RTEvent_Supplier_Config_impl::push_event (
+      Components::EventBase * ev
+      ACE_ENV_ARG_DECL)
+    ACE_THROW_SPEC ((
+      CORBA::SystemException))
+  {
+    if (CIAO::debug_level () > 10)
+      {
+        ACE_DEBUG ((LM_DEBUG, "----CIAO::RTEvent_Supplier_Config::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;
+
+    // We need to find the right "proxy_consumer" object from the
+    // proxy_consumer_map_, and then call push on it.
+
+    this->proxy_consumer_->push (events ACE_ENV_ARG_PARAMETER);
+    ACE_CHECK;
+  }
+
   //////////////////////////////////////////////////////////////////////
   ///                   Consumer Config Implementation
   //////////////////////////////////////////////////////////////////////
@@ -524,10 +597,13 @@ namespace CIAO
     RtecEventComm::EventSourceID source_id =
       hasher (this->supplier_id_.c_str ());
 
+    /* NOTE: This code needs to be refactored to facilitate the current
+             flexibility of event channel connection setup.
     this->qos_.start_disjunction_group (1);
     this->qos_.insert (source_id,
                        source_id,
                        0);
+                       */
   }
 
   void
diff --git a/TAO/CIAO/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.h b/TAO/CIAO/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.h
index 0ba65234b21..e6c6d3031b7 100644
--- a/TAO/CIAO/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.h
+++ b/TAO/CIAO/ciaosvcs/Events/CIAO_RTEC/CIAO_RTEvent.h
@@ -143,6 +143,17 @@ namespace CIAO
     /**
      * @var ACE_Hash_Map_Manager<> proxy_supplier_map_
      *
+     * Mapping of each event publisher to a proxy consumer for disconnect purposes.
+     */
+    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_
+     *
      * Mapping of each event sink to a proxy supplier for disconnect purposes.
      */
     ACE_Hash_Map_Manager_Ex<ACE_CString,
@@ -317,6 +328,18 @@ namespace CIAO
     virtual void destroy (ACE_ENV_SINGLE_ARG_DECL)
       ACE_THROW_SPEC ((CORBA::SystemException));
 
+    virtual void register_proxy_consumer (
+        ::CORBA::Object_ptr proxy_consumer
+        ACE_ENV_ARG_DECL_WITH_DEFAULTS
+      )
+      ACE_THROW_SPEC ((::CORBA::SystemException));
+
+    virtual void push_event (
+        Components::EventBase * ev
+        ACE_ENV_ARG_DECL)
+      ACE_THROW_SPEC ((
+        CORBA::SystemException));
+
   private:
     ACE_CString supplier_id_;
 
@@ -325,6 +348,8 @@ namespace CIAO
     ACE_SupplierQOS_Factory qos_;
 
     PortableServer::POA_var poa_;
+
+    RtecEventChannelAdmin::ProxyPushConsumer_var proxy_consumer_;
   };
 }