A few changes to the new EC implementation

6 files changed, 177 insertions, 14 deletions

diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_ConsumerAdmin.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_ConsumerAdmin.cpp
index 3aaec148f74..448933229fd 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_ConsumerAdmin.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_ConsumerAdmin.cpp
@@ -25,6 +25,48 @@ TAO_EC_ConsumerAdmin::set_default_POA (PortableServer::POA_ptr poa)
     PortableServer::POA::_duplicate (poa);
 }
 
+void
+TAO_EC_ConsumerAdmin::connected (TAO_EC_ProxyPushConsumer *consumer,
+				 CORBA::Environment &ACE_TRY_ENV)
+{
+  SupplierSetIterator end = this->all_suppliers_.end ();
+  for (SupplierSetIterator i = this->all_suppliers_.begin ();
+       i != end;
+       ++i)
+    {
+      (*i)->connected (consumer, ACE_TRY_ENV);
+    }
+}
+
+void
+TAO_EC_ConsumerAdmin::disconnected (TAO_EC_ProxyPushConsumer *consumer,
+				    CORBA::Environment &ACE_TRY_ENV)
+{
+  SupplierSetIterator end = this->all_suppliers_.end ();
+  for (SupplierSetIterator i = this->all_suppliers_.begin ();
+       i != end;
+       ++i)
+    {
+      (*i)->disconnected (consumer, ACE_TRY_ENV);
+    }
+}
+
+void
+TAO_EC_ConsumerAdmin::connected (TAO_EC_ProxyPushSupplier *supplier,
+				 CORBA::Environment &ACE_TRY_ENV)
+{
+  if (this->all_suppliers_.insert (supplier) != 0)
+    ACE_THROW (CORBA::NO_MEMORY (CORBA::COMPLETED_NO));
+}
+
+void
+TAO_EC_ConsumerAdmin::disconnected (TAO_EC_ProxyPushSupplier *supplier,
+				    CORBA::Environment &ACE_TRY_ENV)
+{
+  if (this->all_suppliers_.remove (supplier) != 0)
+    ACE_THROW (RtecEventChannelAdmin::SUBSCRIPTION_ERROR ());
+}
+
 RtecEventChannelAdmin::ProxyPushSupplier_ptr
 TAO_EC_ConsumerAdmin::obtain_push_supplier (CORBA::Environment &ACE_TRY_ENV)
 {
@@ -34,9 +76,6 @@ TAO_EC_ConsumerAdmin::obtain_push_supplier (CORBA::Environment &ACE_TRY_ENV)
   supplier->set_default_POA (
         this->event_channel_->factory ()->supplier_POA ());
 
-  if (this->all_suppliers_.insert (supplier) != 0)
-    ACE_THROW (CORBA::NO_MEMORY (CORBA::COMPLETED_NO));
-
   return supplier->_this (ACE_TRY_ENV);
 }
 
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.cpp
index 759e8a91a6d..a365a5325db 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.cpp
@@ -82,6 +82,46 @@ TAO_EC_Event_Channel::shutdown (CORBA::Environment& ACE_TRY_ENV)
   ACE_CHECK;
 }
 
+void
+TAO_EC_Event_Channel::connected (TAO_EC_ProxyPushConsumer* consumer,
+				 CORBA::Environment &ACE_TRY_ENV)
+{
+  this->consumer_admin_->connected (consumer, ACE_TRY_ENV);
+  ACE_CHECK;
+  this->supplier_admin_->connected (consumer, ACE_TRY_ENV);
+  ACE_CHECK;
+}
+
+void
+TAO_EC_Event_Channel::disconnected (TAO_EC_ProxyPushConsumer* consumer,
+				    CORBA::Environment &ACE_TRY_ENV)
+{
+  this->consumer_admin_->disconnected (consumer, ACE_TRY_ENV);
+  ACE_CHECK;
+  this->supplier_admin_->disconnected (consumer, ACE_TRY_ENV);
+  ACE_CHECK;
+}
+
+void
+TAO_EC_Event_Channel::connected (TAO_EC_ProxyPushSupplier* supplier,
+				 CORBA::Environment &ACE_TRY_ENV)
+{
+  this->supplier_admin_->connected (supplier, ACE_TRY_ENV);
+  ACE_CHECK;
+  this->consumer_admin_->connected (supplier, ACE_TRY_ENV);
+  ACE_CHECK;
+}
+
+void
+TAO_EC_Event_Channel::disconnected (TAO_EC_ProxyPushSupplier* supplier,
+				    CORBA::Environment &ACE_TRY_ENV)
+{
+  this->supplier_admin_->disconnected (supplier, ACE_TRY_ENV);
+  ACE_CHECK;
+  this->consumer_admin_->disconnected (supplier, ACE_TRY_ENV);
+  ACE_CHECK;
+}
+
 RtecEventChannelAdmin::ConsumerAdmin_ptr
 TAO_EC_Event_Channel::for_consumers (CORBA::Environment& ACE_TRY_ENV)
 {
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.i b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.i
index c488a9384cb..dd7c5aa4ae8 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.i
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.i
@@ -30,4 +30,26 @@ TAO_EC_Event_Channel::timer_module (void) const
   this->timer_module_;
 }
 
+ACE_INLINE TAO_EC_ProxyPushSupplier*
+TAO_EC_Event_Channel::create_proxy_push_supplier (void)
+{
+  return this->factory_->create_proxy_push_supplier (this);
+}
+
+ACE_INLINE void
+TAO_EC_Event_Channel::destroy_proxy_push_supplier (TAO_EC_ProxyPushSupplier* supplier)
+{
+  this->factory_->destroy_proxy_push_supplier (supplier);
+}
 
+ACE_INLINE TAO_EC_ProxyPushConsumer*
+TAO_EC_Event_Channel::create_proxy_push_consumer (void)
+{
+  return this->factory_->create_proxy_push_consumer (this);
+}
+
+ACE_INLINE void
+TAO_EC_Event_Channel::destroy_proxy_push_consumer (TAO_EC_ProxyPushConsumer* consumer)
+{
+  this->factory_->destroy_proxy_push_consumer (consumer);
+}
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_ProxyConsumer.h b/TAO/orbsvcs/orbsvcs/Event/EC_ProxyConsumer.h
index 98cb9863612..1fd9463178c 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_ProxyConsumer.h
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_ProxyConsumer.h
@@ -76,11 +76,20 @@ public:
   const RtecEventChannelAdmin::SupplierQOS& publications (void) const;
   // The QoS (subscription) used to connect to the EC.
 
-  virtual void connected (TAO_EC_ProxyPushSupplier* supplier) = 0;
-  virtual void disconnected (TAO_EC_ProxyPushSupplier* supplier) = 0;
+  virtual void connected (TAO_EC_ProxyPushSupplier* supplier,
+			  CORBA::Environment &env) = 0;
+  virtual void disconnected (TAO_EC_ProxyPushSupplier* supplier,
+			     CORBA::Environment &env) = 0;
   // Concrete implementations can use this methods to keep track of
   // the consumers interested in this events.
 
+  virtual void connected (TAO_EC_ProxyPushConsumer* consumer,
+			  CORBA::Environment &env);
+  virtual void disconnected (TAO_EC_ProxyPushConsumer* consumer,
+			     CORBA::Environment &env);
+  // Usually implemented as no-ops, but some configurations may
+  // require this methods.
+
   void set_default_poa (PortableServer::POA_ptr poa);
   // Set this servant's default POA
 
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.h b/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.h
index b63f47bf00a..ce652106ad0 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.h
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.h
@@ -84,6 +84,20 @@ public:
   const RtecEventChannelAdmin::ConsumerQOS& subscriptions (void) const;
   // The QoS (subscription) used to connect to the EC.
 
+  virtual void connected (TAO_EC_ProxyPushConsumer* consumer,
+			  CORBA::Environment &env);
+  virtual void disconnected (TAO_EC_ProxyPushConsumer* consumer,
+			     CORBA::Environment &env);
+  // Concrete implementations can use this methods to keep track of
+  // the suppliers that publish its events.
+
+  virtual void connected (TAO_EC_ProxyPushSupplier* supplier,
+			  CORBA::Environment &env) = 0;
+  virtual void disconnected (TAO_EC_ProxyPushSupplier* supplier,
+			     CORBA::Environment &env) = 0;
+  // Usually implemented as no-ops, but some configurations may
+  // require this methods.
+
   void set_default_poa (PortableServer::POA_ptr poa);
   // Set this servant's default POA
 
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_SupplierAdmin.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_SupplierAdmin.cpp
index 5c21ef3b66e..1ab0d31c9d1 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_SupplierAdmin.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_SupplierAdmin.cpp
@@ -25,6 +25,54 @@ TAO_EC_SupplierAdmin::set_default_POA (PortableServer::POA_ptr poa)
     PortableServer::POA::_duplicate (poa);
 }
 
+PortableServer::POA_ptr
+TAO_EC_SupplierAdmin::_default_POA (CORBA::Environment&)
+{
+  return PortableServer::POA::_duplicate (this->default_POA_.in ());
+}
+
+void
+TAO_EC_SupplierAdmin::connected (TAO_EC_ProxyPushSupplier *supplier,
+				 CORBA::Environment &ACE_TRY_ENV)
+{
+  ConsumerSetIterator end = this->all_consumers_.end ();
+  for (ConsumerSetIterator i = this->all_consumers_.begin ();
+       i != end;
+       ++i)
+    {
+      (*i)->connected (supplier, ACE_TRY_ENV);
+    }
+}
+
+void
+TAO_EC_SupplierAdmin::disconnected (TAO_EC_ProxyPushSupplier *supplier,
+				    CORBA::Environment &ACE_TRY_ENV)
+{
+  ConsumerSetIterator end = this->all_consumers_.end ();
+  for (ConsumerSetIterator i = this->all_consumers_.begin ();
+       i != end;
+       ++i)
+    {
+      (*i)->disconnected (supplier, ACE_TRY_ENV);
+    }
+}
+
+void
+TAO_EC_SupplierAdmin::connected (TAO_EC_ProxyPushConsumer *consumer,
+				 CORBA::Environment &ACE_TRY_ENV)
+{
+  if (this->all_consumers_.insert (consumer) != 0)
+    ACE_THROW (CORBA::NO_MEMORY (CORBA::COMPLETED_NO));
+}
+
+void
+TAO_EC_SupplierAdmin::disconnected (TAO_EC_ProxyPushConsumer *consumer,
+				    CORBA::Environment &ACE_TRY_ENV)
+{
+  if (this->all_consumers_.remove (consumer) != 0)
+    ACE_THROW (RtecEventChannelAdmin::SUBSCRIPTION_ERROR ());
+}
+
 RtecEventChannelAdmin::ProxyPushSupplier_ptr
 TAO_EC_SupplierAdmin::obtain_push_supplier (CORBA::Environment &ACE_TRY_ENV)
 {
@@ -34,14 +82,5 @@ TAO_EC_SupplierAdmin::obtain_push_supplier (CORBA::Environment &ACE_TRY_ENV)
   supplier->set_default_POA (
         this->event_channel_->factory ()->consumer_POA ());
 
-  if (this->all_consumers_.insert (supplier) != 0)
-    ACE_THROW (CORBA::NO_MEMORY (CORBA::COMPLETED_NO));
-
   return supplier->_this (ACE_TRY_ENV);
 }
-
-PortableServer::POA_ptr
-TAO_EC_SupplierAdmin::_default_POA (CORBA::Environment&)
-{
-  return PortableServer::POA::_duplicate (this->default_POA_.in ());
-}