QPID-3170: correct deletion of federation routes when keys match.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1090266 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 376 insertions, 51 deletions

diff --git a/qpid/cpp/src/qpid/broker/DirectExchange.cpp b/qpid/cpp/src/qpid/broker/DirectExchange.cpp
index 5b8104c77c..060f80f60d 100644
--- a/qpid/cpp/src/qpid/broker/DirectExchange.cpp
+++ b/qpid/cpp/src/qpid/broker/DirectExchange.cpp
@@ -94,7 +94,7 @@ bool DirectExchange::bind(Queue::shared_ptr queue, const string& routingKey, con
 
         propagate = bk.fedBinding.delOrigin(queue->getName(), fedOrigin);
         if (bk.fedBinding.countFedBindings(queue->getName()) == 0)
-            unbind(queue, routingKey, 0);
+            unbind(queue, routingKey, args);
 
     } else if (fedOp == fedOpReorigin) {
         /** gather up all the keys that need rebinding in a local vector
@@ -124,17 +124,18 @@ bool DirectExchange::bind(Queue::shared_ptr queue, const string& routingKey, con
     return true;
 }
 
-bool DirectExchange::unbind(Queue::shared_ptr queue, const string& routingKey, const FieldTable* /*args*/)
+bool DirectExchange::unbind(Queue::shared_ptr queue, const string& routingKey, const FieldTable* args)
 {
+    string fedOrigin(args ? args->getAsString(qpidFedOrigin) : "");
     bool propagate = false;
 
-    QPID_LOG(debug, "Unbind key [" << routingKey << "] from queue " << queue->getName());
-
+    QPID_LOG(debug, "Unbinding key [" << routingKey << "] from queue " << queue->getName()
+             << " on exchange " << getName() << " origin=" << fedOrigin << ")" );
     {
         Mutex::ScopedLock l(lock);
         BoundKey& bk = bindings[routingKey];
         if (bk.queues.remove_if(MatchQueue(queue))) {
-            propagate = bk.fedBinding.delOrigin();
+            propagate = bk.fedBinding.delOrigin(queue->getName(), fedOrigin);
             if (mgmtExchange != 0) {
                 mgmtExchange->dec_bindingCount();
             }
diff --git a/qpid/cpp/src/qpid/broker/Exchange.h b/qpid/cpp/src/qpid/broker/Exchange.h
index 9c4e6be192..7199162ddc 100644
--- a/qpid/cpp/src/qpid/broker/Exchange.h
+++ b/qpid/cpp/src/qpid/broker/Exchange.h
@@ -133,15 +133,15 @@ protected:
 
         /** Returns true if propagation is needed. */
         bool delOrigin(const std::string& queueName, const std::string& origin){
-            fedBindings[queueName].erase(origin);
-            return true;
-        }
-
-        /** Returns true if propagation is needed. */
-        bool delOrigin() {
-            if (localBindings > 0)
-                localBindings--;
-            return localBindings == 0;
+            if (origin.empty()) {   // no remote == local binding
+                if (localBindings > 0)
+                    localBindings--;
+                return localBindings == 0;
+            }
+            size_t match = fedBindings[queueName].erase(origin);
+            if (fedBindings[queueName].empty())
+                fedBindings.erase(queueName);
+            return match != 0;
         }
 
         uint32_t count() {
@@ -149,7 +149,11 @@ protected:
         }
 
         uint32_t countFedBindings(const std::string& queueName) {
-            return  fedBindings[queueName].size();
+            // don't use '[]' - it may increase size of fedBindings!
+            std::map<std::string, originSet>::iterator i;
+            if ((i = fedBindings.find(queueName)) != fedBindings.end())
+                return  i->second.size();
+            return 0;
         }
     };
 
diff --git a/qpid/cpp/src/qpid/broker/FanOutExchange.cpp b/qpid/cpp/src/qpid/broker/FanOutExchange.cpp
index ac2c914a97..5879fa0892 100644
--- a/qpid/cpp/src/qpid/broker/FanOutExchange.cpp
+++ b/qpid/cpp/src/qpid/broker/FanOutExchange.cpp
@@ -18,6 +18,7 @@
  * under the License.
  *
  */
+#include "qpid/log/Statement.h"
 #include "qpid/broker/FanOutExchange.h"
 #include "qpid/broker/FedOps.h"
 #include <algorithm>
@@ -65,7 +66,7 @@ bool FanOutExchange::bind(Queue::shared_ptr queue, const string& /*key*/, const
     } else if (fedOp == fedOpUnbind) {
         propagate = fedBinding.delOrigin(queue->getName(), fedOrigin);
         if (fedBinding.countFedBindings(queue->getName()) == 0)
-            unbind(queue, "", 0);
+            unbind(queue, "", args);
     } else if (fedOp == fedOpReorigin) {
         if (fedBinding.hasLocal()) {
             propagateFedOp(string(), string(), fedOpBind, string());
@@ -78,12 +79,16 @@ bool FanOutExchange::bind(Queue::shared_ptr queue, const string& /*key*/, const
     return true;
 }
 
-bool FanOutExchange::unbind(Queue::shared_ptr queue, const string& /*key*/, const FieldTable* /*args*/)
+bool FanOutExchange::unbind(Queue::shared_ptr queue, const string& /*key*/, const FieldTable* args)
 {
+    string fedOrigin(args ? args->getAsString(qpidFedOrigin) : "");
     bool propagate = false;
 
+    QPID_LOG(debug, "Unbinding queue " << queue->getName()
+             << " from exchange " << getName() << " origin=" << fedOrigin << ")" );
+
     if (bindings.remove_if(MatchQueue(queue))) {
-        propagate = fedBinding.delOrigin();
+        propagate = fedBinding.delOrigin(queue->getName(), fedOrigin);
         if (mgmtExchange != 0) {
             mgmtExchange->dec_bindingCount();
         }
diff --git a/qpid/cpp/src/qpid/broker/HeadersExchange.cpp b/qpid/cpp/src/qpid/broker/HeadersExchange.cpp
index 82ac5911ee..abcaa5f69d 100644
--- a/qpid/cpp/src/qpid/broker/HeadersExchange.cpp
+++ b/qpid/cpp/src/qpid/broker/HeadersExchange.cpp
@@ -158,12 +158,13 @@ bool HeadersExchange::bind(Queue::shared_ptr queue, const string& bindingKey, co
     return true;
 }
 
-bool HeadersExchange::unbind(Queue::shared_ptr queue, const string& bindingKey, const FieldTable*){
+bool HeadersExchange::unbind(Queue::shared_ptr queue, const string& bindingKey, const FieldTable *args){
     bool propagate = false;
+    string fedOrigin(args ? args->getAsString(qpidFedOrigin) : "");
     {
         Mutex::ScopedLock l(lock);
 
-        FedUnbindModifier modifier;
+        FedUnbindModifier modifier(queue->getName(), fedOrigin);
         MatchKey match_key(queue, bindingKey);
         bindings.modify_if(match_key, modifier);
         propagate = modifier.shouldPropagate;
@@ -330,11 +331,7 @@ HeadersExchange::FedUnbindModifier::FedUnbindModifier() : shouldUnbind(false), s
 
 bool HeadersExchange::FedUnbindModifier::operator()(BoundKey & bk)
 {
-    if ("" == fedOrigin) {
-        shouldPropagate = bk.fedBinding.delOrigin();
-    } else {
-        shouldPropagate = bk.fedBinding.delOrigin(queueName, fedOrigin);
-    }
+    shouldPropagate = bk.fedBinding.delOrigin(queueName, fedOrigin);
     if (bk.fedBinding.countFedBindings(queueName) == 0)
     {
         shouldUnbind = true;
diff --git a/qpid/cpp/src/qpid/broker/TopicExchange.cpp b/qpid/cpp/src/qpid/broker/TopicExchange.cpp
index 4447790766..644a3d628e 100644
--- a/qpid/cpp/src/qpid/broker/TopicExchange.cpp
+++ b/qpid/cpp/src/qpid/broker/TopicExchange.cpp
@@ -250,21 +250,21 @@ bool TopicExchange::bind(Queue::shared_ptr queue, const string& routingKey, cons
             if (mgmtExchange != 0) {
                 mgmtExchange->inc_bindingCount();
             }
-            QPID_LOG(debug, "Bound key [" << routingPattern << "] to queue " << queue->getName()
-                     << " (origin=" << fedOrigin << ")");
+            QPID_LOG(debug, "Binding key [" << routingPattern << "] to queue " << queue->getName()
+                     << " on exchange " << getName() << " (origin=" << fedOrigin << ")");
         }
     } else if (fedOp == fedOpUnbind) {
-        bool reallyUnbind = false;
-        {
-            RWlock::ScopedWlock l(lock);
-            BindingKey* bk = bindingTree.getBindingKey(routingPattern);
-            if (bk) {
-                propagate = bk->fedBinding.delOrigin(queue->getName(), fedOrigin);
-                reallyUnbind = bk->fedBinding.countFedBindings(queue->getName()) == 0;
+        RWlock::ScopedWlock l(lock);
+        BindingKey* bk = getQueueBinding(queue, routingPattern);
+        if (bk) {
+            QPID_LOG(debug, "FedOpUnbind [" << routingPattern << "] from exchange " << getName()
+                     << " on queue=" << queue->getName() << " origin=" << fedOrigin);
+            propagate = bk->fedBinding.delOrigin(queue->getName(), fedOrigin);
+            // if this was the last binding for the queue, delete the binding
+            if (bk->fedBinding.countFedBindings(queue->getName()) == 0) {
+                deleteBinding(queue, routingPattern, bk);
             }
         }
-        if (reallyUnbind)
-            unbind(queue, routingPattern, 0);
     } else if (fedOp == fedOpReorigin) {
         /** gather up all the keys that need rebinding in a local vector
          * while holding the lock.  Then propagate once the lock is
@@ -289,15 +289,31 @@ bool TopicExchange::bind(Queue::shared_ptr queue, const string& routingKey, cons
     return true;
 }
 
-bool TopicExchange::unbind(Queue::shared_ptr queue, const string& constRoutingKey, const FieldTable* /*args*/){
+bool TopicExchange::unbind(Queue::shared_ptr queue, const string& constRoutingKey, const FieldTable* args)
+{
+    string fedOrigin(args ? args->getAsString(qpidFedOrigin) : "");
+    QPID_LOG(debug, "Unbinding key [" << constRoutingKey << "] from queue " << queue->getName()
+             << " on exchange " << getName() << " origin=" << fedOrigin << ")" );
+
     ClearCache cc(&cacheLock,&bindingCache); // clear the cache on function exit.
     RWlock::ScopedWlock l(lock);
     string routingKey = normalize(constRoutingKey);
-    BindingKey* bk = bindingTree.getBindingKey(routingKey);
+    BindingKey* bk = getQueueBinding(queue, routingKey);
     if (!bk) return false;
-    Binding::vector& qv(bk->bindingVector);
-    bool propagate = false;
+    bool propagate = bk->fedBinding.delOrigin(queue->getName(), fedOrigin);
+    deleteBinding(queue, routingKey, bk);
+    if (propagate)
+        propagateFedOp(routingKey, string(), fedOpUnbind, string());
+    return true;
+}
+
 
+bool TopicExchange::deleteBinding(Queue::shared_ptr queue,
+                                  const std::string& routingKey,
+                                  BindingKey *bk)
+{
+    // Note well: write lock held by caller
+    Binding::vector& qv(bk->bindingVector);
     Binding::vector::iterator q;
     for (q = qv.begin(); q != qv.end(); q++)
         if ((*q)->queue == queue)
@@ -306,31 +322,32 @@ bool TopicExchange::unbind(Queue::shared_ptr queue, const string& constRoutingKe
     qv.erase(q);
     assert(nBindings > 0);
     nBindings--;
-    propagate = bk->fedBinding.delOrigin();
+
     if(qv.empty()) {
         bindingTree.removeBindingKey(routingKey);
     }
     if (mgmtExchange != 0) {
         mgmtExchange->dec_bindingCount();
     }
-    QPID_LOG(debug, "Unbound [" << routingKey << "] from queue " << queue->getName());
-
-    if (propagate)
-        propagateFedOp(routingKey, string(), fedOpUnbind, string());
+    QPID_LOG(debug, "Unbound key [" << routingKey << "] from queue " << queue->getName()
+             << " on exchange " << getName());
     return true;
 }
 
-bool TopicExchange::isBound(Queue::shared_ptr queue, const string& pattern)
+/** returns a pointer to the BindingKey if the given queue is bound to this
+ * exchange using the routing pattern. 0 if queue binding does not exist.
+ */
+TopicExchange::BindingKey *TopicExchange::getQueueBinding(Queue::shared_ptr queue, const string& pattern)
 {
     // Note well: lock held by caller....
     BindingKey *bk = bindingTree.getBindingKey(pattern);  // Exact match against binding pattern
-    if (!bk) return false;
+    if (!bk) return 0;
     Binding::vector& qv(bk->bindingVector);
     Binding::vector::iterator q;
     for (q = qv.begin(); q != qv.end(); q++)
         if ((*q)->queue == queue)
             break;
-    return q != qv.end();
+    return (q != qv.end()) ? bk : 0;
 }
 
 void TopicExchange::route(Deliverable& msg, const string& routingKey, const FieldTable* /*args*/)
@@ -363,7 +380,7 @@ bool TopicExchange::isBound(Queue::shared_ptr queue, const string* const routing
     RWlock::ScopedRlock l(lock);
     if (routingKey && queue) {
         string key(normalize(*routingKey));
-        return isBound(queue, key);
+        return getQueueBinding(queue, key) != 0;
     } else if (!routingKey && !queue) {
         return nBindings > 0;
     } else if (routingKey) {
diff --git a/qpid/cpp/src/qpid/broker/TopicExchange.h b/qpid/cpp/src/qpid/broker/TopicExchange.h
index 79a9b0b021..3109e3b7dd 100644
--- a/qpid/cpp/src/qpid/broker/TopicExchange.h
+++ b/qpid/cpp/src/qpid/broker/TopicExchange.h
@@ -156,7 +156,10 @@ class TopicExchange : public virtual Exchange {
 	     clearCache();
         };
     };
-    bool isBound(Queue::shared_ptr queue, const std::string& pattern);
+    BindingKey *getQueueBinding(Queue::shared_ptr queue, const std::string& pattern);
+    bool deleteBinding(Queue::shared_ptr queue,
+                       const std::string& routingKey,
+                       BindingKey *bk);
 
     class ReOriginIter;
     class BindingsFinderIter;
diff --git a/qpid/cpp/src/tests/federation.py b/qpid/cpp/src/tests/federation.py
index 973a1d366c..201b06a4a2 100755
--- a/qpid/cpp/src/tests/federation.py
+++ b/qpid/cpp/src/tests/federation.py
@@ -23,7 +23,7 @@ from qpid.testlib import TestBase010
 from qpid.datatypes import Message
 from qpid.queue import Empty
 from qpid.util import URL
-from time import sleep
+from time import sleep, time
 
 
 class _FedBroker(object):
@@ -1791,3 +1791,301 @@ class FederationTests(TestBase010):
         if headers:
             return headers[name]
         return None
+
+    def test_dynamic_topic_bounce(self):
+        """ Bounce the connection between federated Topic Exchanges.
+        """
+        class Params:
+            def exchange_type(self): return "topic"
+            def bind_queue(self, ssn, qname, ename):
+                ssn.exchange_bind(queue=qname, exchange=ename,
+                                  binding_key="spud.*")
+            def unbind_queue(self, ssn, qname, ename):
+                ssn.exchange_unbind(queue=qname, exchange=ename, binding_key="spud.*")
+            def delivery_properties(self, ssn):
+                return  ssn.delivery_properties(routing_key="spud.boy")
+
+        self.generic_dynamic_bounce_test(Params())
+
+    def test_dynamic_direct_bounce(self):
+        """ Bounce the connection between federated Direct Exchanges.
+        """
+        class Params:
+            def exchange_type(self): return "direct"
+            def bind_queue(self, ssn, qname, ename):
+                ssn.exchange_bind(queue=qname, exchange=ename, binding_key="spud")
+            def unbind_queue(self, ssn, qname, ename):
+                ssn.exchange_unbind(queue=qname, exchange=ename, binding_key="spud")
+            def delivery_properties(self, ssn):
+                return  ssn.delivery_properties(routing_key="spud")
+        self.generic_dynamic_bounce_test(Params())
+
+    def test_dynamic_fanout_bounce(self):
+        """ Bounce the connection between federated Fanout Exchanges.
+        """
+        class Params:
+            def exchange_type(self): return "fanout"
+            def bind_queue(self, ssn, qname, ename):
+                ssn.exchange_bind(queue=qname, exchange=ename)
+            def unbind_queue(self, ssn, qname, ename):
+                ssn.exchange_unbind(queue=qname, exchange=ename)
+            def delivery_properties(self, ssn):
+                return  ssn.delivery_properties(routing_key="spud")
+        self.generic_dynamic_bounce_test(Params())
+
+    def test_dynamic_headers_bounce(self):
+        """ Bounce the connection between federated Headers Exchanges.
+        """
+        class Params:
+            def exchange_type(self): return "headers"
+            def bind_queue(self, ssn, qname, ename):
+                ssn.exchange_bind(queue=qname, exchange=ename,
+                                  binding_key="spud", arguments={'x-match':'any', 'class':'first'})
+            def unbind_queue(self, ssn, qname, ename):
+                ssn.exchange_unbind(queue=qname, exchange=ename, binding_key="spud")
+            def delivery_properties(self, ssn):
+                return  ssn.message_properties(application_headers={'class':'first'})
+        ## @todo KAG - re-enable once federation bugs with headers exchanges
+        ## are fixed.
+        #self.generic_dynamic_bounce_test(Params())
+        return
+
+
+    def generic_dynamic_bounce_test(self, params):
+        """ Verify that a federated broker can maintain a binding to a local
+        queue using the same key as a remote binding.  Destroy and reconnect
+        the federation link, and verify routes are restored correctly.
+        See QPID-3170.
+        Topology:
+
+        Queue1 <---"Key"---B0<==[Federated Exchange]==>B1---"Key"--->Queue2
+        """
+        session = self.session
+
+        # create the federation
+
+        self.startQmf()
+        qmf = self.qmf
+
+        self._setup_brokers()
+
+        # create exchange on each broker, and retrieve the corresponding
+        # management object for that exchange
+
+        exchanges=[]
+        for _b in self._brokers[0:2]:
+            _b.client_session.exchange_declare(exchange="fedX", type=params.exchange_type())
+            self.assertEqual(_b.client_session.exchange_query(name="fedX").type,
+                             params.exchange_type(), "exchange_declare failed!")
+            # pull the exchange out of qmf...
+            retries = 0
+            my_exchange = None
+            timeout = time() + 10
+            while my_exchange is None and time() <= timeout:
+                objs = qmf.getObjects(_broker=_b.qmf_broker, _class="exchange")
+                for ooo in objs:
+                    if ooo.name == "fedX":
+                        my_exchange = ooo
+                        break
+            if my_exchange is None:
+                self.fail("QMF failed to find new exchange!")
+            exchanges.append(my_exchange)
+
+        #
+        # on each broker, create a local queue bound to the exchange with the
+        # same key value.
+        #
+
+        self._brokers[0].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True)
+        params.bind_queue(self._brokers[0].client_session, "fedX1", "fedX")
+        self.subscribe(self._brokers[0].client_session, queue="fedX1", destination="f1")
+        queue_0 = self._brokers[0].client_session.incoming("f1")
+
+        self._brokers[1].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True)
+        params.bind_queue(self._brokers[1].client_session, "fedX1", "fedX")
+        self.subscribe(self._brokers[1].client_session, queue="fedX1", destination="f1")
+        queue_1 = self._brokers[1].client_session.incoming("f1")
+
+        # now federate the two brokers
+
+        # connect B0 --> B1
+        result = self._brokers[1].qmf_object.connect(self._brokers[0].host,
+                                                     self._brokers[0].port,
+                                                     False, "PLAIN", "guest", "guest", "tcp")
+        self.assertEqual(result.status, 0)
+
+        # connect B1 --> B0
+        result = self._brokers[0].qmf_object.connect(self._brokers[1].host,
+                                                     self._brokers[1].port,
+                                                     False, "PLAIN", "guest", "guest", "tcp")
+        self.assertEqual(result.status, 0)
+
+        # for each link, bridge the "fedX" exchanges:
+
+        for _l in qmf.getObjects(_class="link"):
+            # print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.getBroker())))
+            result = _l.bridge(False,  # durable
+                                 "fedX",  # src
+                                 "fedX",  # dst
+                                 "",  # key
+                                 "",  # tag
+                                 "",  # excludes
+                                 False, # srcIsQueue
+                                 False, # srcIsLocal
+                                 True,  # dynamic
+                                 0)     # sync
+            self.assertEqual(result.status, 0)
+
+        # wait for all the inter-broker links to become operational
+        operational = False
+        timeout = time() + 10
+        while not operational and time() <= timeout:
+            operational = True
+            for _l in qmf.getObjects(_class="link"):
+                #print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.state)))
+                if _l.state != "Operational":
+                    operational = False
+        self.failUnless(operational, "inter-broker links failed to become operational.")
+
+        # @todo - There is no way to determine when the bridge objects become
+        # active.
+
+        # wait until the binding key has propagated to each broker - each
+        # broker should see 2 bindings (1 local, 1 remote)
+
+        binding_counts = [2, 2]
+        self.assertEqual(len(binding_counts), len(exchanges), "Update Test!")
+        for i in range(2):
+            exchanges[i].update()
+            timeout = time() + 10
+            while exchanges[i].bindingCount < binding_counts[i] and time() <= timeout:
+                exchanges[i].update()
+            self.failUnless(exchanges[i].bindingCount == binding_counts[i])
+
+        # send 10 msgs to B0
+        for i in range(1, 11):
+            # dp = self._brokers[0].client_session.delivery_properties(routing_key=params.routing_key())
+            dp = params.delivery_properties(self._brokers[0].client_session)
+            self._brokers[0].client_session.message_transfer(destination="fedX", message=Message(dp, "Message_trp %d" % i))
+
+        # get exactly 10 msgs on B0's local queue and B1's queue
+        for i in range(1, 11):
+            try:
+                msg = queue_0.get(timeout=5)
+                self.assertEqual("Message_trp %d" % i, msg.body)
+                msg = queue_1.get(timeout=5)
+                self.assertEqual("Message_trp %d" % i, msg.body)
+            except Empty:
+                self.fail("Only got %d msgs - expected 10" % i)
+        try:
+            extra = queue_0.get(timeout=1)
+            self.fail("Got unexpected message in queue_0: " + extra.body)
+        except Empty: None
+
+        try:
+            extra = queue_1.get(timeout=1)
+            self.fail("Got unexpected message in queue_1: " + extra.body)
+        except Empty: None
+
+        #
+        # Tear down the bridges between the two exchanges, then wait
+        # for the bindings to be cleaned up
+        #
+
+        for _b in qmf.getObjects(_class="bridge"):
+            result = _b.close()
+            self.assertEqual(result.status, 0)
+
+        binding_counts = [1, 1]
+        self.assertEqual(len(binding_counts), len(exchanges), "Update Test!")
+        for i in range(2):
+            exchanges[i].update()
+            timeout = time() + 10
+            while exchanges[i].bindingCount != binding_counts[i] and time() <= timeout:
+                exchanges[i].update()
+            self.failUnless(exchanges[i].bindingCount == binding_counts[i])
+
+        #
+        # restore the bridges between the two exchanges, and wait for the
+        # bindings to propagate.
+        #
+
+        for _l in qmf.getObjects(_class="link"):
+            # print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.getBroker())))
+            result = _l.bridge(False,  # durable
+                                 "fedX",  # src
+                                 "fedX",  # dst
+                                 "",  # key
+                                 "",  # tag
+                                 "",  # excludes
+                                 False, # srcIsQueue
+                                 False, # srcIsLocal
+                                 True,  # dynamic
+                                 0)     # sync
+            self.assertEqual(result.status, 0)
+
+        binding_counts = [2, 2]
+        self.assertEqual(len(binding_counts), len(exchanges), "Update Test!")
+        for i in range(2):
+            exchanges[i].update()
+            timeout = time() + 10
+            while exchanges[i].bindingCount != binding_counts[i] and time() <= timeout:
+                exchanges[i].update()
+            self.failUnless(exchanges[i].bindingCount == binding_counts[i])
+
+        #
+        # verify traffic flows correctly
+        #
+
+        for i in range(1, 11):
+            #dp = self._brokers[1].client_session.delivery_properties(routing_key=params.routing_key())
+            dp = params.delivery_properties(self._brokers[1].client_session)
+            self._brokers[1].client_session.message_transfer(destination="fedX", message=Message(dp, "Message_trp %d" % i))
+
+        # get exactly 10 msgs on B0's queue and B1's queue
+        for i in range(1, 11):
+            try:
+                msg = queue_0.get(timeout=5)
+                self.assertEqual("Message_trp %d" % i, msg.body)
+                msg = queue_1.get(timeout=5)
+                self.assertEqual("Message_trp %d" % i, msg.body)
+            except Empty:
+                self.fail("Only got %d msgs - expected 10" % i)
+        try:
+            extra = queue_0.get(timeout=1)
+            self.fail("Got unexpected message in queue_0: " + extra.body)
+        except Empty: None
+
+        try:
+            extra = queue_1.get(timeout=1)
+            self.fail("Got unexpected message in queue_1: " + extra.body)
+        except Empty: None
+
+
+        #
+        # cleanup
+        #
+        params.unbind_queue(self._brokers[0].client_session, "fedX1", "fedX")
+        self._brokers[0].client_session.message_cancel(destination="f1")
+        self._brokers[0].client_session.queue_delete(queue="fedX1")
+
+        params.unbind_queue(self._brokers[1].client_session, "fedX1", "fedX")
+        self._brokers[1].client_session.message_cancel(destination="f1")
+        self._brokers[1].client_session.queue_delete(queue="fedX1")
+
+        for _b in qmf.getObjects(_class="bridge"):
+            result = _b.close()
+            self.assertEqual(result.status, 0)
+
+        for _l in qmf.getObjects(_class="link"):
+            result = _l.close()
+            self.assertEqual(result.status, 0)
+
+        for _b in self._brokers[0:2]:
+            _b.client_session.exchange_delete(exchange="fedX")
+
+        self._teardown_brokers()
+
+        self.verify_cleanup()
+
+