ChangeLogTag: Wed Aug 13 13:54:12 UTC 2003 Johnny Willemsen <jwillemsen@remedy.nl>

13 files changed, 495 insertions, 12 deletions

diff --git a/TAO/docs/ec_options.html b/TAO/docs/ec_options.html
index d3f9cdc1035..caf545eeaae 100644
--- a/TAO/docs/ec_options.html
+++ b/TAO/docs/ec_options.html
@@ -633,7 +633,8 @@ static EC_Gateway_IIOP_Factory "-ECGIIOPConsumerECControl null ....."
             <EM>policy</EM>
           </TD>
           <TD>Select the consumer ec control policy (<EM>null</EM> or
-            <EM>reactive</EM>) to detect and discard broken consumer ec's.
+            <EM>reactive</EM> or <EM>reconnect</EM>) to detect and handle 
+             broken consumer ec's.
           </TD>
         </TR>
         <!-- <TR NAME="ECGIIOPConsumerECControlPeriod"> -->
@@ -641,9 +642,9 @@ static EC_Gateway_IIOP_Factory "-ECGIIOPConsumerECControl null ....."
           <TD><CODE>-ECGIIOPConsumerECControlPeriod</CODE>
             <EM>period</EM>
           </TD>
-          <TD>Set the period (in microseconds) used by the reactive
-            supplier control policy (set <EM>ECGIIOPConsumerECControl</EM> to 
-            <EM>reactive</EM>) to poll the state of the consumer ec.
+          <TD>Set the period (in microseconds) used by the reactive and reconnect
+            consumer ec control policy (set <EM>ECGIIOPConsumerECControl</EM> to 
+            <EM>reactive</EM> or <EM>reconnect</EM>) to poll the state of the consumer ec.
             When set to 0 there is no polling but the
             reactive strategy is in place.
           </TD>
@@ -654,9 +655,9 @@ static EC_Gateway_IIOP_Factory "-ECGIIOPConsumerECControl null ....."
           <TD><CODE>-ECGIIOPConsumerECControlTimeout</CODE>
             <EM>timeout</EM>
           </TD>
-          <TD>Set the timeout period (in microseconds) used by the reactive
+          <TD>Set the timeout period (in microseconds) used by the reactive and reconnect
             consumer ec control policy (set <EM>ECGIIOPConsumerECControl</EM> to 
-            <EM>reactive</EM>) to detect a timeout when polling the state
+            <EM>reactive</EM> or <EM>reconnect</EM>) to detect a timeout when polling the state
             of the consumer ec.
           </TD>
         </TR>
diff --git a/TAO/orbsvcs/Event_Service/Event_Service.cpp b/TAO/orbsvcs/Event_Service/Event_Service.cpp
index b0dc7a47f3c..658a4d20c33 100644
--- a/TAO/orbsvcs/Event_Service/Event_Service.cpp
+++ b/TAO/orbsvcs/Event_Service/Event_Service.cpp
@@ -225,6 +225,15 @@ Event_Service::run (int argc, ACE_TCHAR* argv[])
                                   ACE_ENV_ARG_PARAMETER);
           ACE_TRY_CHECK;
 
+          // Creation of persistentPOA is over. Destroy the Policy objects.
+          for (CORBA::ULong i = 0;
+               i < policies.length ();
+               ++i)
+            {
+              policies[i]->destroy (ACE_ENV_SINGLE_ARG_PARAMETER);
+              ACE_TRY_CHECK;
+            }
+
           if (CORBA::is_nil (persistent_poa.in ()))
             ACE_ERROR_RETURN ((LM_ERROR,
                                " (%P|%t) Unable to initialize the persistent POA.\n"),
diff --git a/TAO/orbsvcs/orbsvcs/Event/ECG_Reactive_ConsumerEC_Control.cpp b/TAO/orbsvcs/orbsvcs/Event/ECG_Reactive_ConsumerEC_Control.cpp
index c51089a36bf..7ff87f3f5b6 100644
--- a/TAO/orbsvcs/orbsvcs/Event/ECG_Reactive_ConsumerEC_Control.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/ECG_Reactive_ConsumerEC_Control.cpp
@@ -15,9 +15,9 @@ ACE_RCSID(Event, ECG_Reactive_ConsumerEventChannelControl, "$Id$")
 
 TAO_ECG_Reactive_ConsumerEC_Control::
      TAO_ECG_Reactive_ConsumerEC_Control (const ACE_Time_Value &rate,
-                                                   const ACE_Time_Value &timeout,
-                                                   TAO_EC_Gateway_IIOP* gateway,
-                                                   CORBA::ORB_ptr orb)
+                                          const ACE_Time_Value &timeout,
+                                          TAO_EC_Gateway_IIOP* gateway,
+                                          CORBA::ORB_ptr orb)
   : rate_ (rate),
     timeout_ (timeout),
     adapter_ (this),
@@ -193,6 +193,7 @@ TAO_ECG_Reactive_ConsumerEC_Control::event_channel_not_exist (
       ACE_DEBUG ((LM_DEBUG,
                   "EC_Reactive_ConsumerControl(%P|%t) - "
                   "channel %x does not exists\n"));
+      gateway->cleanup_consumer_ec ();
 
       gateway->cleanup_consumer_proxies (ACE_ENV_SINGLE_ARG_PARAMETER);
       ACE_TRY_CHECK;
@@ -215,6 +216,8 @@ TAO_ECG_Reactive_ConsumerEC_Control::system_exception (
 {
   ACE_TRY
     {
+      gateway->cleanup_consumer_ec ();
+
       gateway->cleanup_consumer_proxies (ACE_ENV_SINGLE_ARG_PARAMETER);
       ACE_TRY_CHECK;
 
diff --git a/TAO/orbsvcs/orbsvcs/Event/ECG_Reconnect_ConsumerEC_Control.cpp b/TAO/orbsvcs/orbsvcs/Event/ECG_Reconnect_ConsumerEC_Control.cpp
new file mode 100644
index 00000000000..4a1d5a068db
--- /dev/null
+++ b/TAO/orbsvcs/orbsvcs/Event/ECG_Reconnect_ConsumerEC_Control.cpp
@@ -0,0 +1,306 @@
+// $Id$
+
+#include "ECG_Reconnect_ConsumerEC_Control.h"
+#include "EC_Gateway_IIOP.h"
+#include "tao/Messaging/Messaging.h"
+#include "tao/ORB_Core.h"
+#include "ace/Reactor.h"
+
+ACE_RCSID(Event, ECG_Reconnect_ConsumerEventChannelControl, "$Id$")
+
+TAO_ECG_Reconnect_ConsumerEC_Control::
+     TAO_ECG_Reconnect_ConsumerEC_Control (const ACE_Time_Value &rate,
+                                           const ACE_Time_Value &timeout,
+                                           TAO_EC_Gateway_IIOP* gateway,
+                                           CORBA::ORB_ptr orb)
+  : rate_ (rate),
+    timeout_ (timeout),
+    adapter_ (this),
+    gateway_ (gateway),
+    orb_ (CORBA::ORB::_duplicate (orb)),
+    is_consumer_ec_connected_ (1)
+{
+  this->reactor_ =
+    this->orb_->orb_core ()->reactor ();
+}
+
+TAO_ECG_Reconnect_ConsumerEC_Control::~TAO_ECG_Reconnect_ConsumerEC_Control (void)
+{
+}
+
+void
+TAO_ECG_Reconnect_ConsumerEC_Control::try_reconnect (
+      ACE_ENV_SINGLE_ARG_DECL)
+{
+  ACE_TRY
+    {
+      CORBA::Boolean disconnected;
+      CORBA::Boolean non_existent =
+      gateway_->consumer_ec_non_existent (disconnected
+                                           ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+      if (!non_existent)
+        {
+          is_consumer_ec_connected_ = 1;
+          gateway_->reconnect_consumer_ec(ACE_ENV_SINGLE_ARG_PARAMETER);
+        }
+    }
+  ACE_CATCHANY
+    {
+      // Ignore all exceptions
+    }
+  ACE_ENDTRY;
+}
+
+void
+TAO_ECG_Reconnect_ConsumerEC_Control::reconnect (
+      ACE_ENV_SINGLE_ARG_DECL)
+{
+  ACE_TRY
+    {
+      is_consumer_ec_connected_ = 1;
+      gateway_->reconnect_consumer_ec(ACE_ENV_SINGLE_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+    }
+  ACE_CATCHANY
+    {
+      // Ignore all exceptions
+    }
+  ACE_ENDTRY;
+}
+
+void
+TAO_ECG_Reconnect_ConsumerEC_Control::query_eventchannel (
+      ACE_ENV_SINGLE_ARG_DECL)
+{
+  ACE_TRY
+    {
+      if (is_consumer_ec_connected_ == 1)
+        {
+          CORBA::Boolean disconnected;
+          CORBA::Boolean non_existent =
+            gateway_->consumer_ec_non_existent (disconnected
+                                                ACE_ENV_ARG_PARAMETER);
+          ACE_TRY_CHECK;
+          if (non_existent && !disconnected)
+            {
+              this->event_channel_not_exist (gateway_ ACE_ENV_ARG_PARAMETER);
+              ACE_TRY_CHECK;
+            }
+        }
+      else
+        {
+           this->try_reconnect(ACE_ENV_SINGLE_ARG_PARAMETER);
+           ACE_TRY_CHECK;
+        }
+    }
+  ACE_CATCH (CORBA::OBJECT_NOT_EXIST, ex)
+    {
+      this->event_channel_not_exist (gateway_ ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+    }
+  ACE_CATCH (CORBA::TRANSIENT, transient)
+    {
+      // This is TAO's minor code for a failed connection, we may
+      // want to be more lenient in the future..
+      // if (transient.minor () == 0x54410085)
+      this->event_channel_not_exist (gateway_ ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+    }
+  ACE_CATCHANY
+    {
+      // Ignore all exceptions
+    }
+  ACE_ENDTRY;
+}
+
+void
+TAO_ECG_Reconnect_ConsumerEC_Control::handle_timeout (
+      const ACE_Time_Value &,
+      const void *)
+{
+  // NOTE, setting RELATIVE_RT_TIMEOUT_POLICY for the duration of
+  // query_eventchannel () below has greater impact than desired.  For
+  // example, while we are pinging consumers here, a nested upcall,
+  // which requires making remote calls may come into the ORB.  Those
+  // remote calls will be carried out with with
+  // RELATIVE_RT_TIMEOUT_POLICY set here in effect.
+  // @@ TODO: should use Guard to set and reset policies.
+  ACE_TRY_NEW_ENV
+    {
+      // Query the state of the Current object *before* we initiate
+      // the iteration...
+      CORBA::PolicyTypeSeq types;
+      CORBA::PolicyList_var policies =
+        this->policy_current_->get_policy_overrides (types
+                                                      ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      // Change the timeout
+      this->policy_current_->set_policy_overrides (this->policy_list_,
+                                                   CORBA::ADD_OVERRIDE
+                                                    ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      // Query the state of the consumers...
+      this->query_eventchannel (ACE_ENV_SINGLE_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      this->policy_current_->set_policy_overrides (policies.in (),
+                                                   CORBA::SET_OVERRIDE
+                                                    ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+      for (CORBA::ULong i = 0; i != policies->length (); ++i)
+        {
+          policies[i]->destroy (ACE_ENV_SINGLE_ARG_PARAMETER);
+          ACE_TRY_CHECK;
+        }
+    }
+  ACE_CATCHANY
+    {
+      // Ignore all exceptions
+    }
+  ACE_ENDTRY;
+}
+
+int
+TAO_ECG_Reconnect_ConsumerEC_Control::activate (void)
+{
+#if defined (TAO_HAS_CORBA_MESSAGING) && TAO_HAS_CORBA_MESSAGING != 0
+  ACE_TRY_NEW_ENV
+    {
+      // Get the PolicyCurrent object
+      CORBA::Object_var tmp =
+        this->orb_->resolve_initial_references ("PolicyCurrent"
+                                                 ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      this->policy_current_ =
+        CORBA::PolicyCurrent::_narrow (tmp.in ()
+                                        ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      // Timeout for polling state (default = 10 msec)
+      TimeBase::TimeT timeout = timeout_.usec() * 10;
+      CORBA::Any any;
+      any <<= timeout;
+
+      this->policy_list_.length (1);
+      this->policy_list_[0] =
+        this->orb_->create_policy (
+               Messaging::RELATIVE_RT_TIMEOUT_POLICY_TYPE,
+               any
+               ACE_ENV_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+
+      // Only schedule the timer, when the rate is not zero
+      if (this->rate_ != ACE_Time_Value::zero)
+      {
+        // Schedule the timer after these policies has been set, because the
+        // handle_timeout uses these policies, if done in front, the channel
+        // can crash when the timeout expires before initiazation is ready.
+        long id = this->reactor_->schedule_timer (&this->adapter_,
+                                                  0,
+                                                  this->rate_,
+                                                  this->rate_);
+        if (id == -1)
+          return -1;
+      }
+    }
+  ACE_CATCHANY
+    {
+      return -1;
+    }
+  ACE_ENDTRY;
+#endif /* TAO_HAS_CORBA_MESSAGING */
+
+  return 0;
+}
+
+int
+TAO_ECG_Reconnect_ConsumerEC_Control::shutdown (void)
+{
+  int r =
+    this->reactor_->remove_handler (&this->adapter_,
+                                    ACE_Event_Handler::DONT_CALL);
+  this->adapter_.reactor (0);
+  return r;
+}
+
+void
+TAO_ECG_Reconnect_ConsumerEC_Control::event_channel_not_exist (
+      TAO_EC_Gateway_IIOP* gateway
+      ACE_ENV_ARG_DECL)
+{
+  ACE_TRY
+    {
+      //ACE_DEBUG ((LM_DEBUG,
+      //            "ECG_Reconnect_ConsumerControl(%P|%t) - "
+      //            "channel %x does not exists\n"));
+      is_consumer_ec_connected_ = 0;
+      gateway->cleanup_consumer_proxies (ACE_ENV_SINGLE_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+    }
+  ACE_CATCHANY
+    {
+      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION,
+                           "TAO_EC_Reconnect_ConsumerControl::event_channel_not_exist");
+      // Ignore all exceptions..
+    }
+  ACE_ENDTRY;
+}
+
+void
+TAO_ECG_Reconnect_ConsumerEC_Control::system_exception (
+      TAO_EC_Gateway_IIOP* gateway,
+      CORBA::SystemException & /* exception */
+      ACE_ENV_ARG_DECL)
+{
+  ACE_TRY
+    {
+      // The current implementation is very strict, and kicks out a
+      // client on the first system exception. We may
+      // want to be more lenient in the future, for example,
+      // this is TAO's minor code for a failed connection.
+      //
+      // if (CORBA::TRANSIENT::_narrow (&exception) != 0
+      //     && exception->minor () == 0x54410085)
+      //   return;
+
+      // Anything else is serious, including timeouts...
+      //ACE_DEBUG ((LM_DEBUG,
+      //            "ECG_Reconnect_ConsumerControl(%P|%t) - "
+      //            "channel %x does not exists system except\n"));
+      is_consumer_ec_connected_ = 0;
+      gateway->cleanup_consumer_proxies (ACE_ENV_SINGLE_ARG_PARAMETER);
+      ACE_TRY_CHECK;
+    }
+  ACE_CATCHANY
+    {
+      // Ignore all exceptions..
+    }
+  ACE_ENDTRY;
+}
+
+// ****************************************************************
+
+TAO_ECG_Reconnect_ConsumerEC_Control_Adapter::TAO_ECG_Reconnect_ConsumerEC_Control_Adapter (
+      TAO_ECG_Reconnect_ConsumerEC_Control *adaptee)
+  :  adaptee_ (adaptee)
+{
+}
+
+int
+TAO_ECG_Reconnect_ConsumerEC_Control_Adapter::handle_timeout (
+      const ACE_Time_Value &tv,
+      const void *arg)
+{
+  this->adaptee_->handle_timeout (tv, arg);
+  return 0;
+}
+
+#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
+
+#elif defined(ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA)
+
+#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */
diff --git a/TAO/orbsvcs/orbsvcs/Event/ECG_Reconnect_ConsumerEC_Control.h b/TAO/orbsvcs/orbsvcs/Event/ECG_Reconnect_ConsumerEC_Control.h
new file mode 100644
index 00000000000..559fb470266
--- /dev/null
+++ b/TAO/orbsvcs/orbsvcs/Event/ECG_Reconnect_ConsumerEC_Control.h
@@ -0,0 +1,133 @@
+/* -*- C++ -*- */
+/**
+ *  @file   ECG_Reconnect_ConsumerEC_Control.h
+ *
+ *  $Id$
+ *
+ *  @author Johnny Willemsen (jwillemsen@remedy.nl)
+ *
+ */
+
+#ifndef TAO_ECG_RECONNECT_CONSUMEREC_CONTROL_H
+#define TAO_ECG_RECONNECT_CONSUMEREC_CONTROL_H
+#include /**/ "ace/pre.h"
+
+#include "ECG_ConsumerEC_Control.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+#include "tao/corba.h"
+#include "ace/Event_Handler.h"
+
+class TAO_ECG_Reconnect_ConsumerEC_Control;
+class TAO_EC_Gateway_IIOP;
+
+/**
+ * @class TAO_ECG_Reconnect_ConsumerEC_Control_Adapter
+ *
+ * @brief Forwards timeout events to the Reconnect ConsumerControl
+ *
+ * The Reconnect ConsumerControl strategy uses the reactor to
+ * periodically wakeup and verify the state of the consumers
+ * registered with the Event Channel.
+ */
+class TAO_RTEvent_Export TAO_ECG_Reconnect_ConsumerEC_Control_Adapter
+  : public ACE_Event_Handler
+{
+public:
+  /// Constructor
+  TAO_ECG_Reconnect_ConsumerEC_Control_Adapter (TAO_ECG_Reconnect_ConsumerEC_Control *adaptee);
+
+  // = Documented in ACE_Event_Handler.
+  virtual int handle_timeout (const ACE_Time_Value &tv,
+                              const void *arg = 0);
+
+private:
+  /// The adapted object
+  TAO_ECG_Reconnect_ConsumerEC_Control *adaptee_;
+};
+
+/**
+ * @class TAO_ECG_Reconnect_ConsumerEC_Control
+ *
+ * @brief Consumer event channel control
+ *
+ * Defines the interface for the consumer event channel control strategy.
+ * This strategy handles misbehaving or failing consumer event channels.
+ *
+ * This control tries to reconnect to the consumer ec incase of a connection
+ * loss and then does a reconnect when the connection is back again. When also
+ * a restart of the consumer ec must be handled correctly, then the consumer
+ * ec must use a persistent IOR.
+ */
+class TAO_RTEvent_Export TAO_ECG_Reconnect_ConsumerEC_Control
+  : public TAO_ECG_ConsumerEC_Control
+{
+public:
+  /// Constructor.  It does not assume ownership of the @a event_channel
+  /// parameter.
+  TAO_ECG_Reconnect_ConsumerEC_Control (const ACE_Time_Value &rate,
+                                                const ACE_Time_Value &timeout,
+                                                TAO_EC_Gateway_IIOP* gateway,
+                                                CORBA::ORB_ptr orb);
+
+  /// Destructor.
+  virtual ~TAO_ECG_Reconnect_ConsumerEC_Control (void);
+
+  /// Receive the timeout from the adapter
+  void handle_timeout (const ACE_Time_Value &tv,
+                       const void* arg);
+
+  // = Documented in TAO_EC_ConsumerControl
+  virtual int activate (void);
+  virtual int shutdown (void);
+  virtual void event_channel_not_exist (TAO_EC_Gateway_IIOP *gateway
+                                        ACE_ENV_ARG_DECL_NOT_USED);
+  virtual void system_exception (TAO_EC_Gateway_IIOP *gateway,
+                                 CORBA::SystemException &
+                                 ACE_ENV_ARG_DECL_NOT_USED);
+
+private:
+  /// Check if the consumers still exists.  It is a helper method for
+  /// handle_timeout() to isolate the exceptions.
+  void query_eventchannel (ACE_ENV_SINGLE_ARG_DECL);
+
+  /// Look if we can reconnect again.
+  void try_reconnect (ACE_ENV_SINGLE_ARG_DECL);
+
+  /// Reconnect to the consumer ec.
+  void reconnect (ACE_ENV_SINGLE_ARG_DECL);
+
+private:
+  /// The polling rate
+  ACE_Time_Value rate_;
+
+  /// The polling timeout
+  ACE_Time_Value timeout_;
+
+  /// The Adapter for the reactor events
+  TAO_ECG_Reconnect_ConsumerEC_Control_Adapter adapter_;
+
+  /// The gateway
+  TAO_EC_Gateway_IIOP* gateway_;
+
+  /// The ORB
+  CORBA::ORB_var orb_;
+
+  /// To control the timeout policy in the thread
+  CORBA::PolicyCurrent_var policy_current_;
+
+  /// Precomputed policy list to the set timeout.
+  CORBA::PolicyList policy_list_;
+
+  /// The ORB reactor
+  ACE_Reactor *reactor_;
+
+  /// Do we have a connection to the consumer ec
+  int is_consumer_ec_connected_;
+};
+
+#include "ace/post.h"
+#endif /* TAO_ECG_RECONNECT_CONSUMEREC_CONTROL_H */
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP.cpp
index 8292126436e..df9d25a77dc 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP.cpp
@@ -168,7 +168,7 @@ TAO_EC_Gateway_IIOP::disconnect_supplier_proxy_i (ACE_ENV_SINGLE_ARG_DECL)
 }
 
 void
-TAO_EC_Gateway_IIOP::reconnect_consumer_ec()
+TAO_EC_Gateway_IIOP::reconnect_consumer_ec(ACE_ENV_SINGLE_ARG_DECL)
 {
   ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
 
@@ -244,7 +244,7 @@ TAO_EC_Gateway_IIOP::update_consumer_i (
 
   // ACE_DEBUG ((LM_DEBUG, "ECG (%t) update_consumer_i \n"));
 
-  this->open_i (c_qos ACE_ENV_SINGLE_ARG_PARAMETER);
+  this->open_i (c_qos ACE_ENV_ARG_PARAMETER);
   ACE_CHECK;
 }
 
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP.h b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP.h
index 2e8c2ae281c..55b72425425 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP.h
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP.h
@@ -94,7 +94,7 @@ public:
       ACE_THROW_SPEC ((CORBA::SystemException));
 
   // Let the gateway reconnect itself to the consumer ec given exisiting QoS
-  void reconnect_consumer_ec();
+  void reconnect_consumer_ec(ACE_ENV_SINGLE_ARG_DECL);
 
   /// Check whether the consumer event channel is non existent or not
   CORBA::Boolean consumer_ec_non_existent (CORBA::Boolean_out disconnected
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP_Factory.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP_Factory.cpp
index c9240341239..b8ab253472c 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP_Factory.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Gateway_IIOP_Factory.cpp
@@ -9,6 +9,7 @@
 #include "ace/Arg_Shifter.h"
 #include "ECG_ConsumerEC_Control.h"
 #include "ECG_Reactive_ConsumerEC_Control.h"
+#include "ECG_Reconnect_ConsumerEC_Control.h"
 #include "EC_Gateway_IIOP.h"
 
 #if !defined (__ACE_INLINE__)
@@ -66,6 +67,8 @@ TAO_EC_Gateway_IIOP_Factory::init (int argc, char* argv[])
                 this->consumer_ec_control_ = 0;
               else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("reactive")) == 0)
                 this->consumer_ec_control_ = 1;
+              else if (ACE_OS::strcasecmp (opt, ACE_LIB_TEXT("reconnect")) == 0)
+                this->consumer_ec_control_ = 2;
               else
                 this->unsupported_option_value ("-ECGIIOPConsumerECControl", opt);
               arg_shifter.consume_arg ();
@@ -173,6 +176,15 @@ TAO_EC_Gateway_IIOP_Factory::create_consumerec_control (TAO_EC_Gateway_IIOP* gat
       ACE_Time_Value rate (0, this->consumer_ec_control_period_);
       return new TAO_ECG_Reactive_ConsumerEC_Control (rate, consumer_ec_control_timeout_, gateway, orb.in ());
     }
+  else if (this->consumer_ec_control_ == 2)
+    {
+      int argc = 0;
+      char **argv = 0;
+      CORBA::ORB_var orb =
+        CORBA::ORB_init (argc, argv, this->orbid_.c_str ());
+      ACE_Time_Value rate (0, this->consumer_ec_control_period_);
+      return new TAO_ECG_Reconnect_ConsumerEC_Control (rate, consumer_ec_control_timeout_, gateway, orb.in ());
+    }
   return 0;
 }
 
diff --git a/TAO/orbsvcs/orbsvcs/Makefile.RTEvent b/TAO/orbsvcs/orbsvcs/Makefile.RTEvent
index 0f668e3fb60..1208058d063 100644
--- a/TAO/orbsvcs/orbsvcs/Makefile.RTEvent
+++ b/TAO/orbsvcs/orbsvcs/Makefile.RTEvent
@@ -94,6 +94,7 @@ CPP_SRCS += \
         Event/EC_Gateway_IIOP_Factory \
         Event/ECG_ConsumerEC_Control \
         Event/ECG_Reactive_ConsumerEC_Control \
+        Event/ECG_Reconnect_ConsumerEC_Control \
         Event/ECG_UDP_Out_Endpoint \
         Event/ECG_UDP_Sender \
         Event/EC_UDP_Admin \
diff --git a/TAO/orbsvcs/orbsvcs/RTEvent.bor b/TAO/orbsvcs/orbsvcs/RTEvent.bor
index 1f4780e4912..960746c8951 100644
--- a/TAO/orbsvcs/orbsvcs/RTEvent.bor
+++ b/TAO/orbsvcs/orbsvcs/RTEvent.bor
@@ -47,6 +47,7 @@ OBJFILES = \
    $(OBJDIR)\EC_Gateway_IIOP_Factory.obj \
    $(OBJDIR)\ECG_ConsumerEC_Control.obj \
    $(OBJDIR)\ECG_Reactive_ConsumerEC_Control.obj \
+   $(OBJDIR)\ECG_Reconnect_ConsumerEC_Control.obj \
    $(OBJDIR)\ECG_UDP_Out_Endpoint.obj \
 	$(OBJDIR)\ECG_Mcast_EH.obj \
 	$(OBJDIR)\ECG_UDP_Sender.obj \
diff --git a/TAO/orbsvcs/orbsvcs/RTEvent.dsp b/TAO/orbsvcs/orbsvcs/RTEvent.dsp
index 8ed5b5ec88a..25dc53f6640 100644
--- a/TAO/orbsvcs/orbsvcs/RTEvent.dsp
+++ b/TAO/orbsvcs/orbsvcs/RTEvent.dsp
@@ -260,6 +260,10 @@ SOURCE=.\Event\ECG_Reactive_ConsumerEC_Control.cpp
 # End Source File

 # Begin Source File

 

+SOURCE=.\Event\ECG_Reconnect_ConsumerEC_Control.cpp

+# End Source File

+# Begin Source File

+

 SOURCE=.\Event\EC_Group_Scheduling.cpp

 # End Source File

 # Begin Source File

@@ -573,6 +577,10 @@ SOURCE=.\Event\ECG_Reactive_ConsumerEC_Control.h
 # End Source File

 # Begin Source File

 

+SOURCE=.\Event\ECG_Reconnect_ConsumerEC_Control.h

+# End Source File

+# Begin Source File

+

 SOURCE=.\Event\EC_Group_Scheduling.h

 # End Source File

 # Begin Source File

diff --git a/TAO/orbsvcs/orbsvcs/RTEvent.mpc b/TAO/orbsvcs/orbsvcs/RTEvent.mpc
index 6785204d4a3..fe11c08f28a 100644
--- a/TAO/orbsvcs/orbsvcs/RTEvent.mpc
+++ b/TAO/orbsvcs/orbsvcs/RTEvent.mpc
@@ -42,6 +42,7 @@ project(RTEvent) : orbsvcslib, core, svc_utils, messaging, portableserver {
       Event/ECG_UDP_Sender.cpp
       Event/ECG_ConsumerEC_Control.cpp
       Event/ECG_Reactive_ConsumerEC_Control.cpp
+      Event/ECG_Reconnect_ConsumerEC_Control.cpp
       Event/EC_And_Filter.cpp
       Event/EC_Basic_Factory.cpp
       Event/EC_Basic_Filter_Builder.cpp
diff --git a/TAO/orbsvcs/orbsvcs/RTEvent_Static.dsp b/TAO/orbsvcs/orbsvcs/RTEvent_Static.dsp
index 605c8f09d82..888bc5314ec 100644
--- a/TAO/orbsvcs/orbsvcs/RTEvent_Static.dsp
+++ b/TAO/orbsvcs/orbsvcs/RTEvent_Static.dsp
@@ -799,6 +799,10 @@ SOURCE=.\Event\ECG_Reactive_ConsumerEC_Control.cpp
 # End Source File

 # Begin Source File

 

+SOURCE=.\Event\ECG_Reconnect_ConsumerEC_Control.cpp

+# End Source File

+# Begin Source File

+

 SOURCE=.\Event\EC_Gateway.cpp

 # End Source File

 # Begin Source File

@@ -1115,6 +1119,10 @@ SOURCE=.\Event\ECG_Reactive_ConsumerEC_Control.h
 # End Source File

 # Begin Source File

 

+SOURCE=.\Event\ECG_Reconnect_ConsumerEC_Control.h

+# End Source File

+# Begin Source File

+

 SOURCE=.\Event\EC_Group_Scheduling.h

 # End Source File

 # Begin Source File