ChangeLogTag:Mon May 10 17:21:52 1999 Carlos O'Ryan <coryan@cs.wustl.edu>

5 files changed, 132 insertions, 10 deletions

diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.cpp
index be3f712ea3c..bd221cfb3a7 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.cpp
@@ -16,14 +16,15 @@
 ACE_RCSID(Event, EC_Event_Channel, "$Id$")
 
 TAO_EC_Event_Channel::
-TAO_EC_Event_Channel (PortableServer::POA_ptr supplier_poa,
-                      PortableServer::POA_ptr consumer_poa,
+TAO_EC_Event_Channel (const TAO_EC_Event_Channel_Attributes& attr,
                       TAO_EC_Factory* factory,
                       int own_factory)
-  : supplier_poa_ (PortableServer::POA::_duplicate (supplier_poa)),
-    consumer_poa_ (PortableServer::POA::_duplicate (consumer_poa)),
+  : supplier_poa_ (PortableServer::POA::_duplicate (attr.supplier_poa)),
+    consumer_poa_ (PortableServer::POA::_duplicate (attr.consumer_poa)),
     factory_ (factory),
-    own_factory_ (own_factory)
+    own_factory_ (own_factory),
+    consumer_reconnect_ (attr.consumer_reconnect),
+    supplier_reconnect_ (attr.supplier_reconnect)
 {
   if (this->factory_ == 0)
     {
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.h b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.h
index 5e92925203b..ee9509bd103 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.h
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.h
@@ -35,6 +35,50 @@
 #include "orbsvcs/RtecEventChannelAdminS.h"
 #include "EC_Factory.h"
 
+class TAO_ORBSVCS_Export TAO_EC_Event_Channel_Attributes
+{
+  // = TITLE
+  //   Defines the construction time attributes for the Event
+  //   Channel.
+  //
+  // = DESCRIPTION
+  //   The event channel implementation is controlled by two
+  //   mechanisms:
+  //      The EC_Factory that provides the strategies for the EC
+  //      implementation.
+  //      The EC attributes that define constants and values required
+  //      by the EC construction.
+  //   This class encapsulates those constants and values, providing
+  //   an easy mechanism to extend the attributes without requiring
+  //   changes in the EC constructor.
+  //
+public:
+  TAO_EC_Event_Channel_Attributes (PortableServer::POA_ptr supplier_poa,
+                                   PortableServer::POA_ptr consumer_poa);
+  // The basic constructor.
+  // The attributes listed as arguments are *required* by the EC, and
+  // no appropiate defaults are available for them.
+
+  // Most fields are public, there is no need to protect them, in fact
+  // the user should be able to set any values she wants.
+
+  int consumer_reconnect;
+  int supplier_reconnect;
+  // Can consumers or suppliers invoke connect_push_* multiple times?
+
+  int consumer_admin_busy_hwm;
+  int max_write_delay;
+  // Flags for the Consumer Admin
+
+private:
+  friend class TAO_EC_Event_Channel;
+  // Only the EC can read the private fields.
+
+  PortableServer::POA_ptr supplier_poa;
+  PortableServer::POA_ptr consumer_poa;
+  // The POAs
+};
+
 class TAO_ORBSVCS_Export TAO_EC_Event_Channel : public POA_RtecEventChannelAdmin::EventChannel
 {
   // = TITLE
@@ -49,8 +93,7 @@ class TAO_ORBSVCS_Export TAO_EC_Event_Channel : public POA_RtecEventChannelAdmin
   //   interface to the EC_Factory.
   //
 public:
-  TAO_EC_Event_Channel (PortableServer::POA_ptr supplier_poa,
-                        PortableServer::POA_ptr consumer_poa,
+  TAO_EC_Event_Channel (const TAO_EC_Event_Channel_Attributes& attributes,
                         TAO_EC_Factory* factory = 0,
                         int own_factory = 0);
   // constructor
@@ -134,6 +177,15 @@ public:
   // Used to inform the EC that a Supplier has connected or
   // disconnected from it.
 
+  // Simple flags to control the EC behavior, set by the application
+  // at construction time.
+
+  int consumer_reconnect (void) const;
+  // Can the consumers reconnect to the EC?
+
+  int supplier_reconnect (void) const;
+  // Can the suppliers reconnect to the EC?
+
   // = The RtecEventChannelAdmin::EventChannel methods...
   virtual RtecEventChannelAdmin::ConsumerAdmin_ptr
       for_consumers (CORBA::Environment& env);
@@ -150,7 +202,7 @@ public:
 
   virtual RtecEventChannelAdmin::Observer_Handle
       append_observer (RtecEventChannelAdmin::Observer_ptr,
-		       CORBA::Environment &env)
+                       CORBA::Environment &env)
       TAO_THROW_SPEC ((CORBA::SystemException,
                        RtecEventChannel::EventChannel::SYNCHRONIZATION_ERROR,
                        RtecEventChannel::EventChannel::CANT_APPEND_OBSERVER));
@@ -195,6 +247,10 @@ private:
 
   TAO_EC_ObserverStrategy *observer_strategy_;
   // The observer strategy
+
+  int consumer_reconnect_;
+  int supplier_reconnect_;
+  // Consumer/Supplier reconnection flags
 };
 
 #if defined (__ACE_INLINE__)
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.i b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.i
index 471be7cfd9b..19c1be241ed 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.i
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Event_Channel.i
@@ -1,5 +1,16 @@
 // $Id$
 
+ACE_INLINE
+TAO_EC_Event_Channel_Attributes::
+TAO_EC_Event_Channel_Attributes (PortableServer::POA_ptr s_poa,
+                                 PortableServer::POA_ptr c_poa)
+  :  consumer_reconnect (0),
+     supplier_reconnect (0),
+     supplier_poa (s_poa),
+     consumer_poa (c_poa)
+{
+}
+
 ACE_INLINE TAO_EC_Dispatching*
 TAO_EC_Event_Channel::dispatching (void) const
 {
@@ -131,3 +142,15 @@ TAO_EC_Event_Channel::destroy_supplier_admin_lock (ACE_Lock* x)
 {
   this->factory_->destroy_supplier_admin_lock (x);
 }
+
+ACE_INLINE int
+TAO_EC_Event_Channel::consumer_reconnect (void) const
+{
+  return this->consumer_reconnect_;
+}
+
+ACE_INLINE int
+TAO_EC_Event_Channel::supplier_reconnect (void) const
+{
+  return this->consumer_reconnect_;
+}
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.cpp
index d1384c41d72..9af16c805f7 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.cpp
@@ -96,7 +96,50 @@ TAO_EC_ProxyPushSupplier::connect_push_consumer (
     ACE_CHECK;
 
     if (this->is_connected_i ())
-      ACE_THROW (RtecEventChannelAdmin::AlreadyConnected ());
+      {
+        if (this->event_channel_->consumer_reconnect () == 0)
+          ACE_THROW (RtecEventChannelAdmin::AlreadyConnected ());
+
+        // Re-connections are allowed, go ahead and disconnect the
+        // consumer...
+        this->consumer_ =
+          RtecEventComm::PushConsumer::_nil ();
+
+        // @@ Why don't we have a destroy() method in the
+        // filter_builder?
+        delete this->child_;
+        this->child_ = 0;
+
+        // @@ Are there any race conditions here:
+        //   + The lock is released, but the object is marked as
+        //     disconnected already, so:
+        //     - No events will be pushed
+        //     - Any disconnects will just return
+        //   + But another thread could invoke connect_push_consumer()
+        //     again, notice that by the time the lock is acquired
+        //     again the connected() call may still be running.
+        //     It seems like we need delayed operations again, or
+        //     something similar to what the POA does in this
+        //     scenario.
+        //     Meanwhile we can tell the users: "if it hurts don't do
+        //     it".
+        //
+        TAO_EC_Unlock reverse_lock (*this->lock_);
+
+        {
+          ACE_GUARD_THROW_EX (
+              TAO_EC_Unlock, ace_mon, reverse_lock,
+              RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR ());
+          ACE_CHECK;
+
+          this->event_channel_->disconnected (this, ACE_TRY_ENV);
+          ACE_CHECK;
+        }
+
+        // What if a second thread connected us after this?
+        if (this->is_connected_i ())
+          return;
+      }
 
     this->consumer_ =
       RtecEventComm::PushConsumer::_duplicate (push_consumer);
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Trivial_Supplier_Filter.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Trivial_Supplier_Filter.cpp
index c714690164e..ab7bf213903 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Trivial_Supplier_Filter.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Trivial_Supplier_Filter.cpp
@@ -97,4 +97,3 @@ TAO_EC_Trivial_Supplier_Filter_Builder::destroy (
     TAO_EC_SupplierFiltering*)
 {
 }
-