From c296c9bd6d6090ad9321615d6fffa5372b5e0eb4 Mon Sep 17 00:00:00 2001 From: Kenneth Anthony Giusti Date: Fri, 8 Apr 2011 14:28:43 +0000 Subject: 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 --- qpid/cpp/src/qpid/broker/DirectExchange.cpp | 11 +- qpid/cpp/src/qpid/broker/Exchange.h | 24 ++- qpid/cpp/src/qpid/broker/FanOutExchange.cpp | 11 +- qpid/cpp/src/qpid/broker/HeadersExchange.cpp | 11 +- qpid/cpp/src/qpid/broker/TopicExchange.cpp | 65 +++--- qpid/cpp/src/qpid/broker/TopicExchange.h | 5 +- qpid/cpp/src/tests/federation.py | 300 ++++++++++++++++++++++++++- 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::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 @@ -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() + + -- cgit v1.2.1