ChangeLogTag:Tue Feb 1 11:25:00 2000 Carlos O'Ryan <coryan@uci.edu>

15 files changed, 110 insertions, 59 deletions

diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Basic_Filter_Builder.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Basic_Filter_Builder.cpp
index 046da2bb373..d715dcfdd0d 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Basic_Filter_Builder.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Basic_Filter_Builder.cpp
@@ -49,34 +49,25 @@ TAO_EC_Basic_Filter_Builder:: recursive_build (
       TAO_EC_Filter** children;
       ACE_NEW_RETURN (children, TAO_EC_Filter*[n], 0);
       CORBA::ULong i = 0;
-      for (; i != n && pos != l; ++i)
+      for (; i != n; ++i)
         {
           children[i] = this->recursive_build (supplier, qos, pos);
         }
-      if (pos == l)
-        ++i;
-      return new TAO_EC_Conjunction_Filter (children, i);
+      return new TAO_EC_Conjunction_Filter (children, n);
     }
-  else if (e.header.type == ACE_ES_DISJUNCTION_DESIGNATOR
-           || e.header.type == ACE_ES_GLOBAL_DESIGNATOR)
+  else if (e.header.type == ACE_ES_DISJUNCTION_DESIGNATOR)
     {
       pos++; // Consume the designator
       CORBA::ULong n = this->count_children (qos, pos);
 
       TAO_EC_Filter** children;
       ACE_NEW_RETURN (children, TAO_EC_Filter*[n], 0);
-      for (CORBA::ULong i = 0; i != n; ++i)
-        {
-          children[i] = this->recursive_build (supplier, qos, pos);
-        }
       CORBA::ULong i = 0;
-      for (; i != n && pos != l; ++i)
+      for (; i != n; ++i)
         {
           children[i] = this->recursive_build (supplier, qos, pos);
         }
-      if (pos == l)
-        ++i;
-      return new TAO_EC_Disjunction_Filter (children, i);
+      return new TAO_EC_Disjunction_Filter (children, n);
     }
   else if (e.header.type == ACE_ES_NEGATION_DESIGNATOR)
     {
@@ -152,17 +143,12 @@ TAO_EC_Basic_Filter_Builder::
 {
   CORBA::ULong l = qos.dependencies.length ();
   CORBA::ULong i;
-
-  int n = 0;
-  for (i = pos; i < l; ++i)
+  for (i = pos; i != l; ++i)
     {
       const RtecEventComm::Event& e = qos.dependencies[i].event;
       if (e.header.type == ACE_ES_CONJUNCTION_DESIGNATOR
           || e.header.type == ACE_ES_DISJUNCTION_DESIGNATOR)
         break;
-      // @@ Properly count the number of children, and not the number
-      // of nodes below...
-      ++n;
     }
-  return n;
+  return i - pos;
 }
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_ConsumerAdmin.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_ConsumerAdmin.cpp
index 284845dfca1..c8beaaed225 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_ConsumerAdmin.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_ConsumerAdmin.cpp
@@ -76,7 +76,7 @@ void
 TAO_EC_ConsumerAdmin::reconnected (TAO_EC_ProxyPushSupplier *supplier,
                                    CORBA::Environment &ACE_TRY_ENV)
 {
-  this->collection_->connected (supplier, ACE_TRY_ENV);
+  this->collection_->reconnected (supplier, ACE_TRY_ENV);
 }
 
 void
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Defaults.h b/TAO/orbsvcs/orbsvcs/Event/EC_Defaults.h
index ed2872ecad8..3801e567bab 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Defaults.h
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Defaults.h
@@ -29,6 +29,10 @@
 #ifndef TAO_EC_DEFAULTS_H
 #define TAO_EC_DEFAULTS_H
 
+#ifndef TAO_EC_ENABLE_DEBUG_MESSAGES
+#define TAO_EC_ENABLE_DEBUG_MESSAGES 0
+#endif /* TAO_EC_ENABLE_DEBUG_MESSAGES */
+
 // Any of the definitions below can be overriden in the config.h file
 // or the compilation line.
 
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Per_Supplier_Filter.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Per_Supplier_Filter.cpp
index 8cf9e21d8de..94f456de3ab 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Per_Supplier_Filter.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Per_Supplier_Filter.cpp
@@ -86,15 +86,24 @@ TAO_EC_Per_Supplier_Filter::connected (TAO_EC_ProxyPushSupplier* supplier,
       const RtecEventComm::Event& event =
         pub.publications[j].event;
 
-      //      ACE_DEBUG ((LM_DEBUG, "Trying %d:%d in %x\n",
-      //                  event.header.source, event.header.type,
-      //                  this));
+#if TAO_EC_ENABLE_DEBUG_MESSAGES
+      ACE_DEBUG ((LM_DEBUG, "Connecting consumer <%x> to <%x>, "
+                  "trying event <%d:%d>  ",
+                  supplier, this,
+                  event.header.source, event.header.type));
+#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
       if (supplier->can_match (event.header))
         {
-          //          ACE_DEBUG ((LM_DEBUG, "  matched %x\n", supplier));
+#if TAO_EC_ENABLE_DEBUG_MESSAGES
+          ACE_DEBUG ((LM_DEBUG, "  matched\n"));
+#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
           this->collection_->connected (supplier, ACE_TRY_ENV);
+          ACE_CHECK;
           return;
         }
+#if TAO_EC_ENABLE_DEBUG_MESSAGES
+      ACE_DEBUG ((LM_DEBUG, "  not matched\n"));
+#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
     }
 }
 
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_ProxyConsumer.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_ProxyConsumer.cpp
index 19e80ef110e..d1653c11c40 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_ProxyConsumer.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_ProxyConsumer.cpp
@@ -296,6 +296,10 @@ TAO_EC_ProxyPushConsumer::connect_push_supplier (
       RtecEventComm::PushSupplier::_duplicate (push_supplier);
     this->qos_ = qos;
 
+#if TAO_EC_ENABLE_DEBUG_MESSAGES
+    ACE_DEBUG ((LM_DEBUG, "Building filter for supplier <%x>\n",
+                this));
+#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
     this->filter_ =
       this->event_channel_->supplier_filter_builder ()->create (this->qos_);
     this->filter_->bind (this);
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.cpp
index 6bbde98f08e..383634884f1 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_ProxySupplier.cpp
@@ -264,6 +264,10 @@ TAO_EC_ProxyPushSupplier::connect_push_consumer (
       RtecEventComm::PushConsumer::_duplicate (push_consumer);
     this->qos_ = qos;
 
+#if TAO_EC_ENABLE_DEBUG_MESSAGES
+    ACE_DEBUG ((LM_DEBUG, "Building filters for consumer <%x>\n",
+                this));
+#endif /* TAO_EC_ENABLED_DEBUG_MESSAGES */
     this->child_ =
       this->event_channel_->filter_builder ()->build (this,
                                                       this->qos_,
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Supplier_Filter.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Supplier_Filter.cpp
index 01d4c7ae1be..54573aa5844 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Supplier_Filter.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Supplier_Filter.cpp
@@ -21,6 +21,5 @@ TAO_EC_Filter_Worker::work (TAO_EC_ProxyPushSupplier *supplier,
                             CORBA::Environment &ACE_TRY_ENV)
 {
   TAO_EC_QOS_Info qos_info = this->event_info_;
-
   supplier->filter (this->event_, qos_info, ACE_TRY_ENV);
 }
diff --git a/TAO/orbsvcs/orbsvcs/Event/EC_Type_Filter.cpp b/TAO/orbsvcs/orbsvcs/Event/EC_Type_Filter.cpp
index 1d289a70159..814c0b56b47 100644
--- a/TAO/orbsvcs/orbsvcs/Event/EC_Type_Filter.cpp
+++ b/TAO/orbsvcs/orbsvcs/Event/EC_Type_Filter.cpp
@@ -96,6 +96,18 @@ TAO_EC_Type_Filter::can_match (
         return this->header_.source == header.source;
     }
 
+  if (header.source == 0)
+    {
+      if (header.type != 0)
+        return this->header_.type == header.type;
+      return 1;
+    }
+
+  if (header.type == 0)
+    {
+      return this->header_.source == header.source;
+    }
+
   return (this->header_.type == header.type
           && this->header_.source == header.source);
 }
diff --git a/TAO/orbsvcs/tests/EC_Custom_Marshal/ECM_Supplier.cpp b/TAO/orbsvcs/tests/EC_Custom_Marshal/ECM_Supplier.cpp
index c40148860fc..51de95aeb54 100644
--- a/TAO/orbsvcs/tests/EC_Custom_Marshal/ECM_Supplier.cpp
+++ b/TAO/orbsvcs/tests/EC_Custom_Marshal/ECM_Supplier.cpp
@@ -462,7 +462,7 @@ Test_Supplier::disconnect (CORBA::Environment &ACE_TRY_ENV)
   RtecEventChannelAdmin::ProxyPushConsumer_var proxy =
     this->consumer_proxy_._retn ();
 
-  this->consumer_proxy_->disconnect_push_consumer (ACE_TRY_ENV);
+  proxy->disconnect_push_consumer (ACE_TRY_ENV);
 }
 
 int
diff --git a/TAO/orbsvcs/tests/EC_Throughput/ec.conf b/TAO/orbsvcs/tests/EC_Throughput/ec.conf
index 9327a35965c..ecec16431a2 100644
--- a/TAO/orbsvcs/tests/EC_Throughput/ec.conf
+++ b/TAO/orbsvcs/tests/EC_Throughput/ec.conf
@@ -1,2 +1,2 @@
 # $Id$
-static EC_Factory "-ECdispatching reactive -ECfiltering basic -ECproxyconsumerlock thread -ECsupplieradminlock thread -ECsupplierfiltering per-supplier"
+static EC_Factory "-ECdispatching reactive -ECfiltering basic -ECproxyconsumerlock thread -ECsupplierfiltering per-supplier"
diff --git a/TAO/orbsvcs/tests/Event/Basic/Atomic_Reconnect.cpp b/TAO/orbsvcs/tests/Event/Basic/Atomic_Reconnect.cpp
index 40b80a761bb..a2b44a79b80 100644
--- a/TAO/orbsvcs/tests/Event/Basic/Atomic_Reconnect.cpp
+++ b/TAO/orbsvcs/tests/Event/Basic/Atomic_Reconnect.cpp
@@ -89,7 +89,7 @@ main (int argc, char* argv[])
 
       // ****************************************************************
 
-      Consumer consumer01 ("Consumer/010", event_type);
+      Consumer consumer01 ("Consumer/01", event_type);
       // Create a consumer, intialize its RT_Info structures, and
       // connnect to the event channel....
 
@@ -207,13 +207,18 @@ main (int argc, char* argv[])
       // ****************************************************************
 
       ACE_DEBUG ((LM_DEBUG,
-                  "Supplier 0 pushed %d events\n", task0.push_count ()));
+                  "Task 0 pushed %d events\n", task0.push_count ()));
       ACE_DEBUG ((LM_DEBUG,
-                  "Supplier 1 pushed %d events\n", task1.push_count ()));
+                  "Task 1 pushed %d events\n", task1.push_count ()));
+      ACE_DEBUG ((LM_DEBUG,
+                  "Supplier 0 pushed %d events\n", supplier0.event_count));
+      ACE_DEBUG ((LM_DEBUG,
+                  "Supplier 1 pushed %d events\n", supplier1.event_count));
       consumer0.dump_results (task0.push_count (), 5);
       consumer1.dump_results (task1.push_count (), 5);
-      consumer01.dump_results (task0.push_count (), 1,
-                               task0.push_count () + task1.push_count ());
+      consumer01.dump_results (task0.push_count (),
+                               task1.push_count (),
+                               1);
     }
   ACE_CATCHANY
     {
@@ -242,28 +247,45 @@ Consumer::dump_results (int expected_count,
 }
 
 void
-Consumer::dump_results (int expected_count,
-                        int tolerance,
-                        int max_base_count)
+Consumer::dump_results (int base_count,
+                        int extra_count,
+                        int tolerance)
 {
-  this->dump_results (expected_count, tolerance);
+  int diff = this->event_base_count - base_count;
+  if (diff > tolerance || diff < -tolerance)
+    {
+      ACE_DEBUG ((LM_DEBUG,
+                  "ERROR - %s unexpected number of base events  <%d>\n",
+                  this->name_,
+                  this->event_base_count));
+    }
+  else
+    {
+      ACE_DEBUG ((LM_DEBUG,
+                  "%s - number of base events <%d> within margins\n",
+                  this->name_,
+                  this->event_base_count));
+    }
 
-  if (event_base_count >= CORBA::ULong (max_base_count))
+  if (this->event_count < base_count
+      || this->event_count >= base_count + extra_count)
     {
       ACE_DEBUG ((LM_DEBUG,
-                  "ERROR - %s unexpected number of events  <%d,%d>\n",
+                  "ERROR - %s unexpected number of events  <%d,%d,%d>\n",
                   this->name_,
-                  this->event_base_count,
-                  max_base_count));
+                  base_count,
+                  this->event_count,
+                  base_count + extra_count));
     }
   else
     {
       ACE_DEBUG ((LM_DEBUG,
                   "%s number of events "
-                  "<%d,%d> within margins\n",
+                  "<%d,%d,%d> within margins\n",
                   this->name_,
-                  this->event_base_count,
-                  max_base_count));
+                  base_count,
+                  this->event_count,
+                  base_count + extra_count));
     }
 }
 
@@ -282,6 +304,10 @@ Consumer::push (const RtecEventComm::EventSet& events,
   ACE_GUARD (ACE_SYNCH_MUTEX, ace_mon, this->lock_);
   this->event_count++;
 
+  // ACE_DEBUG ((LM_DEBUG,
+  //               "Consumer %s has received %d events\n",
+  //               this->name_, this->event_count));
+
   if (events[0].header.type == this->event_base_type_)
     this->event_base_count++;
 }
diff --git a/TAO/orbsvcs/tests/Event/Basic/Atomic_Reconnect.h b/TAO/orbsvcs/tests/Event/Basic/Atomic_Reconnect.h
index a8b94997e6c..63179e972b0 100644
--- a/TAO/orbsvcs/tests/Event/Basic/Atomic_Reconnect.h
+++ b/TAO/orbsvcs/tests/Event/Basic/Atomic_Reconnect.h
@@ -38,9 +38,9 @@ public:
 
   void dump_results (int expected_count,
                      int tolerance);
-  void dump_results (int expected_count,
-                     int tolerance,
-                     int max_base_count);
+  void dump_results (int base_count,
+                     int extra_count,
+                     int tolerance);
 
   // = The RtecEventComm::PushConsumer methods
 
diff --git a/TAO/orbsvcs/tests/Event/Basic/svc.conf b/TAO/orbsvcs/tests/Event/Basic/svc.conf
index 75444c8a77c..88d85850ce7 100644
--- a/TAO/orbsvcs/tests/Event/Basic/svc.conf
+++ b/TAO/orbsvcs/tests/Event/Basic/svc.conf
@@ -1,2 +1,2 @@
 # $Id$
-static EC_Factory "-ECProxyPushConsumerCollection mt:delayed:list -ECProxyPushSupplierCollection mt:delayed:list -ECdispatching reactive -ECfiltering basic -ECproxyconsumerlock thread -ECproxysupplierlock thread -ECsupplierfiltering null"
+static EC_Factory "-ECProxyPushConsumerCollection mt:delayed:list -ECProxyPushSupplierCollection mt:delayed:list -ECdispatching reactive -ECfiltering basic -ECproxyconsumerlock thread -ECproxysupplierlock thread -ECsupplierfiltering per-supplier"
diff --git a/TAO/orbsvcs/tests/Event/lib/Counting_Consumer.cpp b/TAO/orbsvcs/tests/Event/lib/Counting_Consumer.cpp
index 24b610b5dde..aa34043b720 100644
--- a/TAO/orbsvcs/tests/Event/lib/Counting_Consumer.cpp
+++ b/TAO/orbsvcs/tests/Event/lib/Counting_Consumer.cpp
@@ -22,9 +22,12 @@ EC_Counting_Consumer::connect (RtecEventChannelAdmin::ConsumerAdmin_ptr consumer
     this->_this (ACE_TRY_ENV);
   ACE_CHECK;
 
-  this->supplier_proxy_ =
-    consumer_admin->obtain_push_supplier (ACE_TRY_ENV);
-  ACE_CHECK;
+  if (CORBA::is_nil (this->supplier_proxy_.in ()))
+    {
+      this->supplier_proxy_ =
+        consumer_admin->obtain_push_supplier (ACE_TRY_ENV);
+      ACE_CHECK;
+    }
 
   this->supplier_proxy_->connect_push_consumer (consumer.in (),
                                                 qos,
@@ -46,6 +49,9 @@ EC_Counting_Consumer::disconnect (CORBA::Environment &ACE_TRY_ENV)
   ACE_CHECK;
   consumer_poa->deactivate_object (consumer_id.in (), ACE_TRY_ENV);
   ACE_CHECK;
+
+  this->supplier_proxy_ =
+    RtecEventChannelAdmin::ProxyPushSupplier::_nil ();
 }
 
 void
diff --git a/TAO/orbsvcs/tests/Event/lib/Counting_Supplier.cpp b/TAO/orbsvcs/tests/Event/lib/Counting_Supplier.cpp
index 53ae0ba4d62..bbc4021b5f5 100644
--- a/TAO/orbsvcs/tests/Event/lib/Counting_Supplier.cpp
+++ b/TAO/orbsvcs/tests/Event/lib/Counting_Supplier.cpp
@@ -64,10 +64,10 @@ EC_Counting_Supplier::deactivate (CORBA::Environment &ACE_TRY_ENV)
 void
 EC_Counting_Supplier::connect (
     RtecEventChannelAdmin::SupplierAdmin_ptr supplier_admin,
-    int event_source,
-    int event_type,
     int published_source,
     int published_type,
+    int event_source,
+    int event_type,
     CORBA::Environment &ACE_TRY_ENV)
 {
   this->event_source_ = event_source;
@@ -92,9 +92,12 @@ EC_Counting_Supplier::connect (
     this->_this (ACE_TRY_ENV);
   ACE_CHECK;
 
-  this->consumer_proxy_ =
-    supplier_admin->obtain_push_consumer (ACE_TRY_ENV);
-  ACE_CHECK;
+  if (CORBA::is_nil (this->supplier_proxy_.in ()))
+    {
+      this->consumer_proxy_ =
+        supplier_admin->obtain_push_consumer (ACE_TRY_ENV);
+      ACE_CHECK;
+    }
 
   this->consumer_proxy_->connect_push_supplier (supplier.in (),
                                                 qos,
@@ -195,8 +198,6 @@ void
 EC_Counting_Supplier_Task::run (CORBA::Environment &ACE_TRY_ENV)
 {
   this->event_.length (1);
-  this->event_[0].header.type = 0;
-  this->event_[0].header.source = 0;
 
   int stop = 0;
   do {