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

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

6 files changed, 77 insertions, 50 deletions

diff --git a/cpp/src/qpid/broker/DirectExchange.cpp b/cpp/src/qpid/broker/DirectExchange.cpp
index 5b8104c77c..060f80f60d 100644
--- a/cpp/src/qpid/broker/DirectExchange.cpp
+++ b/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/cpp/src/qpid/broker/Exchange.h b/cpp/src/qpid/broker/Exchange.h
index 9c4e6be192..7199162ddc 100644
--- a/cpp/src/qpid/broker/Exchange.h
+++ b/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/cpp/src/qpid/broker/FanOutExchange.cpp b/cpp/src/qpid/broker/FanOutExchange.cpp
index ac2c914a97..5879fa0892 100644
--- a/cpp/src/qpid/broker/FanOutExchange.cpp
+++ b/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/cpp/src/qpid/broker/HeadersExchange.cpp b/cpp/src/qpid/broker/HeadersExchange.cpp
index 82ac5911ee..abcaa5f69d 100644
--- a/cpp/src/qpid/broker/HeadersExchange.cpp
+++ b/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/cpp/src/qpid/broker/TopicExchange.cpp b/cpp/src/qpid/broker/TopicExchange.cpp
index 4447790766..644a3d628e 100644
--- a/cpp/src/qpid/broker/TopicExchange.cpp
+++ b/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/cpp/src/qpid/broker/TopicExchange.h b/cpp/src/qpid/broker/TopicExchange.h
index 79a9b0b021..3109e3b7dd 100644
--- a/cpp/src/qpid/broker/TopicExchange.h
+++ b/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;