Fully functional, as far as I can tell.

10 files changed, 359 insertions, 106 deletions

diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer.cpp b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer.cpp
index 62358938e78..97da36d9be2 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer.cpp
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer.cpp
@@ -20,8 +20,8 @@
 
 ACE_RCSID(EC_Examples, Consumer, "$Id$")
 
-Consumer::Consumer (RtecEventComm::EventSourceID normal_type,
-                    RtecEventComm::EventSourceID ft_type,
+Consumer::Consumer (SourceID normal_type,
+                    SourceID ft_type,
                     Supplier *fwddest, Service_Handler * handler)
   : worktime_(0,0)
   , fwddest_(fwddest)
@@ -31,8 +31,8 @@ Consumer::Consumer (RtecEventComm::EventSourceID normal_type,
 {
 }
 
-Consumer::Consumer (RtecEventComm::EventSourceID normal_type,
-                    RtecEventComm::EventSourceID ft_type,
+Consumer::Consumer (SourceID normal_type,
+                    SourceID ft_type,
                     ACE_Time_Value& worktime,
                     Supplier *fwddest, Service_Handler *handler)
   : worktime_(worktime)
@@ -43,6 +43,10 @@ Consumer::Consumer (RtecEventComm::EventSourceID normal_type,
 {
 }
 
+Consumer::~Consumer(void)
+{
+}
+
 void
 Consumer::push (const RtecEventComm::EventSet& events
                 ACE_ENV_ARG_DECL)
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer.h b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer.h
index 9a294f68cbc..390fa6611ff 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer.h
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer.h
@@ -46,19 +46,23 @@ class Consumer : public POA_RtecEventComm::PushConsumer
   //   and it is up to the driver program to use the right one.
   //
 public:
+  typedef RtecEventComm::EventSourceID SourceID;
   typedef RtecScheduler::handle_t InfoHandle;
+  typedef RtecEventComm::EventType EventType;
 
-  Consumer (RtecEventComm::EventSourceID normal_type,
-            RtecEventComm::EventSourceID ft_type,
+  Consumer (SourceID normal_type,
+            SourceID ft_type,
             ACE_Time_Value& worktime,
             Supplier *fwddest = 0, Service_Handler *handler = 0);
   // Constructor
 
-  Consumer (RtecEventComm::EventSourceID normal_type,
-            RtecEventComm::EventSourceID ft_type,
+  Consumer (SourceID normal_type,
+            SourceID ft_type,
             Supplier *fwddest = 0, Service_Handler *handler = 0);
   // Constructor
 
+  virtual ~Consumer(void);
+
   // = The RtecEventComm::PushConsumer methods
 
   virtual void push (const RtecEventComm::EventSet& events
@@ -77,7 +81,7 @@ public:
 
   Service_Handler * handler(void) const;
 
-private:
+protected:
   int deadline_missed_;
   ACE_Time_Value worktime_;
 
@@ -87,8 +91,8 @@ private:
 
   Service_Handler * handler_;
 
-  RtecEventComm::EventType norm_type_;
-  RtecEventComm::EventType ft_type_;
+  EventType norm_type_;
+  EventType ft_type_;
 };
 
 #endif /* CONSUMER_H */
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_EC.cpp b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_EC.cpp
index 5ff34d73328..d010243aee9 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_EC.cpp
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Consumer_EC.cpp
@@ -17,7 +17,7 @@
 #include "tao/ORB_Core.h"
 
 #include "Kokyu_EC.h"
-#include "Consumer.h"
+#include "Dynamic_Consumer.h"
 #include "Service_Handler.h"
 
 #ifdef ACE_HAS_DSUI
@@ -34,82 +34,6 @@ namespace
   ACE_CString ior_output_filename;
   FILE * ior_output_file;
 }
-/*
-class Once_Handler: public Service_Handler
-{
-public:
-  Once_Handler(void)
-    : handled_start_(0)
-    , handled_stop_(0)
-  {
-  }
-
-  virtual ~Once_Handler(void)
-  {
-    this->reactor_->cancel_timer(this->timer_handle_);
-
-    delete timeout_handler_impl_;
-  }
-
-  ///Takes ownership of the Timeout_Handler ONLY.
-  void init (
-             ACE_Reactor * reactor
-             , ACE_Event_Handler *timeout_handler_impl
-             , ACE_Time_Value period
-             )
-  {
-    this->reactor_ = reactor;
-    this->timeout_handler_impl_ = timeout_handler_impl;
-    this->period_ = period;
-  }
-
-  virtual void handle_service_start(ACE_ENV_SINGLE_ARG_DECL)
-    ACE_THROW_SPEC ((
-                     CORBA::SystemException
-                     , RtecScheduler::UNKNOWN_TASK
-                     , RtecScheduler::INTERNAL
-                     , RtecScheduler::SYNCHRONIZATION_FAILURE
-                     ))
-  {
-    if (this->handled_start_ == 0)
-      {
-        ACE_DEBUG((LM_DEBUG,"Once_Handler (%P|%t) handle_service_start() START\n"));
-        this->handled_start_++; //set to true
-
-        //WARNING: depending on Reactor, might not be a RT solution!
-
-        this->timer_handle_ = this->reactor_->schedule_timer(this->timeout_handler_impl_,
-                                                             0, //arg
-                                                             ACE_Time_Value::zero, //delay
-                                                             this->period_ //period
-                                                             );
-
-        ACE_DEBUG((LM_DEBUG,"Once_Handler (%P|%t) handle_service_start() END\n"));
-      }
-  }
-
-  virtual void handle_service_stop(const RtecEventComm::EventSet&
-                                    ACE_ENV_ARG_DECL)
-  {
-    if (this->handled_stop_ == 0)
-      {
-        ACE_DEBUG((LM_DEBUG,"Once_Handler (%P|%t) handle_service_stop() START\n"));
-        this->handled_stop_++; //set to true
-        ACE_DEBUG((LM_DEBUG,"Once_Handler (%P|%t) handle_service_stop() END\n"));
-      }
-  }
-
-private:
-  int handled_start_;
-  int handled_stop_;
-
-  ACE_Reactor *reactor_;
-  ACE_Event_Handler *timeout_handler_impl_;
-  ACE_Time_Value period_;
-  long timer_handle_;
-
-};
-*/
 
 class Consumer_EC : public Kokyu_EC
 {
@@ -137,15 +61,20 @@ public:
     //TODO: Create Dynamic_Consumer for merging fault-tolerant case!
 
     //// CONSUMER 3 ////
-    RtecEventComm::EventType cons_normal_type = ACE_ES_EVENT_UNDEFINED+6;
-    RtecEventComm::EventType cons_ft_type = ACE_ES_EVENT_UNDEFINED+8;
+    Consumer::EventType cons_normal_type  = ACE_ES_EVENT_UNDEFINED+6;
+    Consumer::EventType cons_normal_type2 = ACE_ES_EVENT_UNDEFINED+7;
+    Consumer::EventType cons_ft_type  = ACE_ES_EVENT_UNDEFINED+8;
+    Consumer::EventType cons_ft_type2 = ACE_ES_EVENT_UNDEFINED+9;
     Kokyu_EC::EventType_Vector cons1_3_types(2);
     cons1_3_types.push_back(cons_normal_type);
     cons1_3_types.push_back(cons_ft_type);
+    cons1_3_types.push_back(cons_normal_type2);
+    cons1_3_types.push_back(cons_ft_type2);
 
-    Consumer * consumer_impl1_3;
+    Dynamic_Consumer * consumer_impl1_3;
     ACE_NEW(consumer_impl1_3,
-            Consumer(cons_normal_type,cons_ft_type));
+            Dynamic_Consumer(cons_normal_type,cons_ft_type,
+                             cons_normal_type2,cons_ft_type2));
 
     tv.set(0,50000);
     consumer_impl1_3->setWorkTime(tv);
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Consumer.cpp b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Consumer.cpp
new file mode 100644
index 00000000000..e497b6e7b5c
--- /dev/null
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Consumer.cpp
@@ -0,0 +1,242 @@
+// $Id$
+
+#include "Dynamic_Consumer.h"
+#include "Supplier.h"
+#include "Service_Handler.h"
+
+#include "orbsvcs/orbsvcs/Time_Utilities.h" //ORBSVCS_Time
+
+#include "ace/Timeprobe.h"
+#include "ace/High_Res_Timer.h"
+#include "ace/Time_Value.h"
+#include "ace/ACE.h" //for is_prime()
+#include "ace/Counter.h"
+#include "ace/OS_NS_sys_time.h"
+
+#if defined (ACE_HAS_DSUI)
+#include <dsui.h>
+#include "federated_config.h"
+#include "federated_dsui_families.h"
+#endif /* ACE_HAS_DSUI */
+
+ACE_RCSID(EC_Examples, Dynamic_Consumer, "$Id$")
+
+Dynamic_Consumer::Dynamic_Consumer (SourceID normal_type,
+                                    SourceID ft_type,
+                                    SourceID normal_type2,
+                                    SourceID ft_type2,
+                                    ACE_Time_Value& worktime,
+                                    Supplier *fwddest,
+                                    Service_Handler *handler)
+  : Consumer(normal_type,ft_type,worktime,fwddest,handler)
+  , norm_type2_(normal_type2)
+  , ft_type2_(ft_type2)
+  , received_ft_type1_(false)
+  , received_ft_type2_(false)
+{
+}
+
+Dynamic_Consumer::Dynamic_Consumer (SourceID normal_type,
+                                    SourceID ft_type,
+                                    SourceID normal_type2,
+                                    SourceID ft_type2,
+                                    Supplier *fwddest,
+                                    Service_Handler *handler)
+  : Consumer(normal_type,ft_type,fwddest,handler)
+  , norm_type2_(normal_type2)
+  , ft_type2_(ft_type2)
+  , received_ft_type1_(false)
+  , received_ft_type2_(false)
+{
+}
+
+Dynamic_Consumer::~Dynamic_Consumer(void)
+{
+}
+
+void
+Dynamic_Consumer::push (const RtecEventComm::EventSet& events
+                        ACE_ENV_ARG_DECL)
+  ACE_THROW_SPEC ((CORBA::SystemException))
+{
+  if (events.length () == 0)
+    {
+      ACE_DEBUG ((LM_DEBUG,
+                  "Dynamic_Consumer (%P|%t) no events\n"));
+      return;
+    }
+
+  //WARNING: We are assuming only one event is pushed at a time!
+
+  ACE_DEBUG ((LM_DEBUG, "Dynamic_Consumer (%P|%t) we received event type %d\n",
+              events[0].header.type));
+
+  Supplier::mode_t mode;
+  if (events[0].header.type == this->ft_type_
+      || events[0].header.type == this->ft_type2_)
+    {
+      mode = Supplier::FAULT_TOLERANT;
+      if (events[0].header.type == this->ft_type_)
+        {
+          if (this->received_ft_type1_)
+            {
+              ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) received second event (type %d) before receiving type %d!\n",this->ft_type_,this->ft_type2_));
+            }
+          this->received_ft_type1_ = true;
+        }
+      else {
+        //must be ft_type2_
+          if (this->received_ft_type2_)
+            {
+              ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) received second event (type %d) before receiving type %d!\n",this->ft_type2_,this->ft_type_));
+            }
+        this->received_ft_type2_ = true;
+      }
+    }
+  else if (events[0].header.type == this->norm_type_
+      || events[0].header.type == this->norm_type2_)
+    {
+      mode = Supplier::NORMAL;
+      if (this->received_ft_type1_ || this->received_ft_type2_) {
+        ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) mode is NORMAL, but both FAULT_TOLERANT events not received!\n"));
+      }
+    }
+  else
+    {
+      //default
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) received unknown type %d; ignoring\n",events[0].header.type));
+      return;
+    }
+
+  if (this->handler_ != 0)
+    {
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) calling handle_service_start()\n"));
+      this->handler_->handle_service_start(ACE_ENV_SINGLE_ARG_PARAMETER);
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) handle_service_start() DONE\n"));
+    }
+
+  //NOTE: We don't trigger START/STOP_SERVICE events or handle fwddest if
+  //not ready because if we aren't ready, it's just overhead.
+  //We are ready only if we're in NORMAL mode (we should receive only one
+  //event) or if we're in FAULT_TOLERANT mode and have received both events.
+  if (Supplier::NORMAL == mode ||
+      (Supplier::FAULT_TOLERANT == mode &&
+      this->received_ft_type1_ && this->received_ft_type2_))
+    {
+
+      //@BT INSTRUMENT with event ID: EVENT_WORK_START Measure time
+      //when work triggered by event starts.
+      //DSTRM_EVENT (TEST_ONE_FAM, START_SERVICE, guid, 0, NULL);
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) START_SERVICE at %u\n",ACE_OS::gettimeofday().msec()));
+
+      Object_ID oid;
+      oid.id = events[0].header.eid.id;
+      oid.tid = events[0].header.eid.tid;
+      oid.pid = events[0].header.eid.pid;
+      oid.queue_id = events[0].header.eid.queue_id;
+      oid.type = events[0].header.type;
+
+      ACE_TIMEPROBE("START_SERVICE");
+      DSTRM_EVENT (TEST_ONE_FAM, START_SERVICE, 0, sizeof(Object_ID), (char*)&oid);
+
+      ACE_High_Res_Timer timer;
+      ACE_Time_Value elapsed_time;
+
+      static CORBA::ULong prime_number = 9619899;
+
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) worktime is %isec %iusec\n",
+                 this->worktime_.sec(),this->worktime_.usec()));
+
+      //There seems to be something wrong with using ACE_Countdown_Time, so we don't use it.
+      //ACE_Time_Value compute_count_down_time (this->worktime_);
+      //ACE_Countdown_Time compute_count_down (&compute_count_down_time); //auto-starts
+
+      ACE_Time_Value start_time(ACE_OS::gettimeofday());
+      timer.start();
+      int j=0;
+      while (elapsed_time <= this->worktime_)
+        {
+          //ACE_DEBUG((LM_DEBUG,"%isec %iusec elapsed\n",elapsed_time.sec(),elapsed_time.usec()));
+
+          ACE::is_prime (prime_number,
+                         2,
+                         prime_number / 2);
+
+          ++j;
+          elapsed_time = ACE_OS::gettimeofday() - start_time;
+        }
+
+      //ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) elapsed %isec %iusec\n",elapsed_time.sec(),elapsed_time.usec()));
+
+      timer.stop ();
+      timer.elapsed_time (elapsed_time); //total elapsed time
+
+      TimeBase::TimeT now;
+      ORBSVCS_Time::Time_Value_to_TimeT (now, ACE_OS::gettimeofday ());
+      if(now > events[0].header.deadline )
+        {
+          this->deadline_missed_++;
+
+          //@BT INSTRUMENT with event ID: EVENT_WORK_DEADLINE_MISSED Measure time when
+          //work triggered by event finishes and deadline missed.
+          //DSTRM_EVENT (TEST_ONE_FAM, DEADLINE_MISSED, guid, strlen(extra_info), extra_info);
+          ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer in thread %t STOP_SERVICE (DEADLINE_MISSED) at %u\n",ACE_OS::gettimeofday().msec()));
+          DSTRM_EVENT (TEST_ONE_FAM, DEADLINE_MISSED, 0, sizeof(Object_ID), (char*)&oid);
+        }
+
+      //@BT INSTRUMENT with event ID: EVENT_WORK_END Measure time when
+      //work triggered by event finishes.
+      //DSTRM_EVENT (TEST_ONE_FAM, STOP_SERVICE, guid,0,NULL);
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer in thread %t STOP_SERVICE at %u\n",ACE_OS::gettimeofday().msec()));
+      DSTRM_EVENT (TEST_ONE_FAM, STOP_SERVICE, 0, sizeof(Object_ID), (char*)&oid);
+
+      ACE_TIMEPROBE("STOP_SERVICE");
+
+      //now print timeprobe values
+      ACE_TIMEPROBE_PRINT;
+      ACE_TIMEPROBE_RESET;
+
+      ACE_DEBUG ((LM_DEBUG, "Dynamic_Consumer (%P|%t) request processing for %d done, "
+                  "elapsed time = %isec %iusec, deadline_missed_=%d\n",
+                  events[0].header.type,elapsed_time.sec(),elapsed_time.usec(),
+                  this->deadline_missed_));
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) processing took %d iterations\n",j));
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) event had deadline %i\n",
+                 events[0].header.deadline));
+
+      //   ACE_DEBUG ((LM_DEBUG,
+      //               "Request processing in thread %t done, "
+      //               "prio = %d, load = %d, elapsed time = %umsec, deadline_missed = %d\n",
+      //               prio, exec_duration, elapsed_time.msec (),Deadline_missed ));
+
+      //now, trigger the next subtask if any
+      if (this->fwddest_ != 0)
+        {
+          //trigger next subtask; we assume we are the only ones who set the Supplier's mode!
+          ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) triggering next subtask\n"));
+          this->fwddest_->mode(mode);
+          this->fwddest_->timeout_occured(ACE_ENV_SINGLE_ARG_PARAMETER);
+        }
+
+      //get ready for next event
+      this->received_ft_type1_ = false;
+      this->received_ft_type2_ = false;
+    } //else not ready to run
+  else
+    {
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) NOT READY to run\n"));
+    }
+
+  if (this->handler_ != 0)
+    {
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) calling handle_service_stop()\n"));
+      this->handler_->handle_service_stop(ACE_ENV_SINGLE_ARG_PARAMETER);
+      ACE_DEBUG((LM_DEBUG,"Dynamic_Consumer (%P|%t) handle_service_stop() DONE\n"));
+    }
+}
+
+// ****************************************************************
+
+#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/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Consumer.h b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Consumer.h
new file mode 100644
index 00000000000..4e0147e9173
--- /dev/null
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Consumer.h
@@ -0,0 +1,76 @@
+/* -*- C++ -*- */
+// $Id$
+//
+// ============================================================================
+//
+// = LIBRARY
+//   ORBSVCS Real-time Event Channel examples
+//
+// = FILENAME
+//   Dynamic_Consumer
+//
+// = AUTHOR
+//   Bryan Thrall
+//
+// ============================================================================
+
+#ifndef DYNAMIC_CONSUMER_H
+#define DYNAMIC_CONSUMER_H
+
+#include "Consumer.h"
+#include "Supplier.h"
+#include "Service_Handler.h"
+
+#if !defined (ACE_LACKS_PRAGMA_ONCE)
+# pragma once
+#endif /* ACE_LACKS_PRAGMA_ONCE */
+
+class Dynamic_Consumer : public Consumer
+{
+public:
+  Dynamic_Consumer (SourceID normal_type,
+                    SourceID ft_type,
+                    SourceID normal_type2,
+                    SourceID ft_type2,
+                    ACE_Time_Value& worktime,
+                    Supplier *fwddest = 0, Service_Handler *handler = 0);
+  // Constructor
+
+  Dynamic_Consumer (SourceID normal_type,
+                    SourceID ft_type,
+                    SourceID normal_type2,
+                    SourceID ft_type2,
+                    Supplier *fwddest = 0, Service_Handler *handler = 0);
+  // Constructor
+
+  virtual ~Dynamic_Consumer(void);
+
+  // = The RtecEventComm::PushConsumer methods
+
+  virtual void push (const RtecEventComm::EventSet& events
+                     ACE_ENV_ARG_DECL)
+    ACE_THROW_SPEC ((CORBA::SystemException));
+
+private:
+  /*
+  int deadline_missed_;
+  ACE_Time_Value worktime_;
+
+  Supplier *fwddest_;
+  //RtecScheduler::handle_t rt_info_;
+  InfoHandle rt_info_;
+
+  Service_Handler * handler_;
+
+  EventType norm_type_;
+  EventType ft_type_;
+  */
+
+  EventType norm_type2_;
+  EventType ft_type2_;
+
+  bool received_ft_type1_;
+  bool received_ft_type2_;
+};
+
+#endif /* DYNAMIC_CONSUMER_H */
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Supplier.cpp b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Supplier.cpp
index eb6fdcfb5c6..6c95658e952 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Supplier.cpp
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Supplier.cpp
@@ -85,19 +85,21 @@ Dynamic_Supplier::timeout_occured (ACE_ENV_SINGLE_ARG_DECL)
   ACE_DEBUG((LM_DEBUG,"Dynamic_Supplier (id %d) in thread %t ONE_WAY_CALL_START at %u\n",this->id_,ACE_OS::gettimeofday().msec()));
   oid.type = eventA[0].header.type;
   DSTRM_EVENT (WORKER_GROUP_FAM, ONE_WAY_CALL_START, 0, sizeof(Object_ID), (char*)&oid);
-  //TODO: BUG? This code pushes eventA/B to ALL consumers!
 
   ACE_DEBUG((LM_DEBUG,"Dynamic_Supplier (id %d) in thread %t pushing eventA %d\n",this->id_,eventA[0].header.type));
   this->consumer_proxy_->push (eventA ACE_ENV_ARG_PARAMETER);
   DSTRM_EVENT (WORKER_GROUP_FAM, ONE_WAY_CALL_DONE, 0, sizeof(Object_ID), (char*)&oid);
 
-  oid.type = eventB[0].header.type;
-  DSTRM_EVENT (WORKER_GROUP_FAM, ONE_WAY_CALL_START, 0, sizeof(Object_ID), (char*)&oid);
+  if (FAULT_TOLERANT == this->mode_) {
+    //only push to both if in FAULT_TOLERANT mode
+    oid.type = eventB[0].header.type;
+    DSTRM_EVENT (WORKER_GROUP_FAM, ONE_WAY_CALL_START, 0, sizeof(Object_ID), (char*)&oid);
 
-  ACE_DEBUG((LM_DEBUG,"Dynamic_Supplier (id %d) in thread %t pushing eventB %d\n",this->id_,eventB[0].header.type));
-  this->consumer_proxy_->push (eventB ACE_ENV_ARG_PARAMETER);
-  ACE_DEBUG((LM_DEBUG,"Dynamic_Supplier (id %d) in thread %t ONE_WAY_CALL_DONE at %u\n",this->id_,ACE_OS::gettimeofday().msec()));
-  DSTRM_EVENT (WORKER_GROUP_FAM, ONE_WAY_CALL_DONE, 0, sizeof(Object_ID), (char*)&oid);
+    ACE_DEBUG((LM_DEBUG,"Dynamic_Supplier (id %d) in thread %t pushing eventB %d\n",this->id_,eventB[0].header.type));
+    this->consumer_proxy_->push (eventB ACE_ENV_ARG_PARAMETER);
+    ACE_DEBUG((LM_DEBUG,"Dynamic_Supplier (id %d) in thread %t ONE_WAY_CALL_DONE at %u\n",this->id_,ACE_OS::gettimeofday().msec()));
+    DSTRM_EVENT (WORKER_GROUP_FAM, ONE_WAY_CALL_DONE, 0, sizeof(Object_ID), (char*)&oid);
+  }
 
   if (this->handler_ != 0)
     {
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Supplier.h b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Supplier.h
index 82e5eb1990b..3a33786d383 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Supplier.h
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Dynamic_Supplier.h
@@ -16,13 +16,7 @@
 
 #ifndef DYNAMIC_SUPPLIER_H
 #define DYNAMIC_SUPPLIER_H
-/*
-#include "ace/Event_Handler.h"
 
-#include "orbsvcs/RtecEventCommS.h"
-#include "orbsvcs/RtecSchedulerC.h"
-#include "orbsvcs/Event/EC_Event_Channel.h"
-*/
 #include "Supplier.h"
 #include "Service_Handler.h"
 
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.h b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.h
index 6186a4eb9b6..e848fee12cd 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.h
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier.h
@@ -21,6 +21,7 @@
 #include "ace/Vector_T.h"
 
 #include "orbsvcs/RtecEventCommS.h"
+#include "orbsvcs/RtecEventChannelAdminC.h"
 #include "orbsvcs/RtecSchedulerC.h"
 #include "orbsvcs/Event/EC_Event_Channel.h"
 
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier_EC.cpp b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier_EC.cpp
index acefc321b5f..1030e9c836a 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier_EC.cpp
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/Supplier_EC.cpp
@@ -225,7 +225,7 @@ public:
 
     Mode_Handler *mode_handler;
     ACE_NEW(mode_handler,
-            Mode_Handler(5)); //mode switch after first event
+            Mode_Handler(1)); //mode switch after first event
     Supplier *supplier_impl1_1;
     Timeout_Consumer *timeout_consumer_impl1_1;
     ACE_NEW(supplier_impl1_1,
diff --git a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/dyn_test.mpc b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/dyn_test.mpc
index 7ddefda3841..38c78b71875 100644
--- a/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/dyn_test.mpc
+++ b/TAO/orbsvcs/examples/RtEC/Federated_Kokyu/dynamic_topology_test/dyn_test.mpc
@@ -9,6 +9,7 @@ project(Consumer_Dynamic): orbsvcsexe, rtevent, rtsched, rtschedevent, rtrgkokyu
     Consumer_EC.cpp
     Consumer.cpp
     Supplier.cpp
+    Dynamic_Consumer.cpp
     federated_dsui_vars.cpp
   }
 }