diff options
Diffstat (limited to 'cpp/src')
-rw-r--r-- | cpp/src/qpid/broker/DirectExchange.cpp | 11 | ||||
-rw-r--r-- | cpp/src/qpid/broker/FanOutExchange.cpp | 2 | ||||
-rw-r--r-- | cpp/src/qpid/broker/HeadersExchange.cpp | 1 | ||||
-rw-r--r-- | cpp/src/qpid/broker/TopicExchange.cpp | 5 |
4 files changed, 18 insertions, 1 deletions
diff --git a/cpp/src/qpid/broker/DirectExchange.cpp b/cpp/src/qpid/broker/DirectExchange.cpp index 05179502e6..1787c01af9 100644 --- a/cpp/src/qpid/broker/DirectExchange.cpp +++ b/cpp/src/qpid/broker/DirectExchange.cpp @@ -76,6 +76,9 @@ bool DirectExchange::bind(Queue::shared_ptr queue, const string& routingKey, con BoundKey& bk = bindings[routingKey]; if (exclusiveBinding) bk.queues.clear(); + QPID_LOG(debug, "Bind key [" << routingKey << "] to queue " << queue->getName() + << " (origin=" << fedOrigin << ")"); + if (bk.queues.add_unless(b, MatchQueue(queue))) { b->startManagement(); propagate = bk.fedBinding.addOrigin(fedOrigin); @@ -83,11 +86,17 @@ bool DirectExchange::bind(Queue::shared_ptr queue, const string& routingKey, con mgmtExchange->inc_bindingCount(); } } else { + // queue already present - still need to track fedOrigin + bk.fedBinding.addOrigin(fedOrigin); return false; } } else if (fedOp == fedOpUnbind) { Mutex::ScopedLock l(lock); BoundKey& bk = bindings[routingKey]; + + QPID_LOG(debug, "Bind - fedOpUnbind key [" << routingKey << "] queue " << queue->getName() + << " (origin=" << fedOrigin << ")"); + propagate = bk.fedBinding.delOrigin(fedOrigin); if (bk.fedBinding.count() == 0) unbind(queue, routingKey, 0); @@ -123,6 +132,8 @@ bool DirectExchange::unbind(Queue::shared_ptr queue, const string& routingKey, c { bool propagate = false; + QPID_LOG(debug, "Unbind key [" << routingKey << "] from queue " << queue->getName()); + { Mutex::ScopedLock l(lock); BoundKey& bk = bindings[routingKey]; diff --git a/cpp/src/qpid/broker/FanOutExchange.cpp b/cpp/src/qpid/broker/FanOutExchange.cpp index ef410a9154..d8e55decbd 100644 --- a/cpp/src/qpid/broker/FanOutExchange.cpp +++ b/cpp/src/qpid/broker/FanOutExchange.cpp @@ -69,6 +69,8 @@ bool FanOutExchange::bind(Queue::shared_ptr queue, const string& /*key*/, const mgmtExchange->inc_bindingCount(); } } else { + // queue already present - still need to track fedOrigin + fedBinding.addOrigin(fedOrigin); return false; } } else if (fedOp == fedOpUnbind) { diff --git a/cpp/src/qpid/broker/HeadersExchange.cpp b/cpp/src/qpid/broker/HeadersExchange.cpp index daf53085bf..6a723b373e 100644 --- a/cpp/src/qpid/broker/HeadersExchange.cpp +++ b/cpp/src/qpid/broker/HeadersExchange.cpp @@ -120,6 +120,7 @@ bool HeadersExchange::bind(Queue::shared_ptr queue, const string& bindingKey, co mgmtExchange->inc_bindingCount(); } } else { + bk.fedBinding.addOrigin(fedOrigin); return false; } } // lock dropped diff --git a/cpp/src/qpid/broker/TopicExchange.cpp b/cpp/src/qpid/broker/TopicExchange.cpp index 66ace42cfa..7372e58c4a 100644 --- a/cpp/src/qpid/broker/TopicExchange.cpp +++ b/cpp/src/qpid/broker/TopicExchange.cpp @@ -205,6 +205,9 @@ bool TopicExchange::bind(Queue::shared_ptr queue, const string& routingKey, cons if (args == 0 || fedOp.empty() || fedOp == fedOpBind) { RWlock::ScopedWlock l(lock); if (isBound(queue, routingPattern)) { + // already bound, but may be from a different fedOrigin + BoundKey& bk = bindings[routingPattern]; + bk.fedBinding.addOrigin(fedOrigin); return false; } else { Binding::shared_ptr binding (new Binding (routingPattern, queue, this, FieldTable(), fedOrigin)); @@ -215,7 +218,7 @@ bool TopicExchange::bind(Queue::shared_ptr queue, const string& routingKey, cons if (mgmtExchange != 0) { mgmtExchange->inc_bindingCount(); } - QPID_LOG(debug, "Bound [" << routingPattern << "] to queue " << queue->getName() + QPID_LOG(debug, "Bound key [" << routingPattern << "] to queue " << queue->getName() << " (origin=" << fedOrigin << ")"); } } else if (fedOp == fedOpUnbind) { |