Add ability for a queue to record all bindings to it, such that these can be removed when the queue is deleted.

Fix to QPID-438



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

6 files changed, 124 insertions, 3 deletions

diff --git a/cpp/src/qpid/broker/BrokerAdapter.cpp b/cpp/src/qpid/broker/BrokerAdapter.cpp
index 592995f10f..e099c4a56d 100644
--- a/cpp/src/qpid/broker/BrokerAdapter.cpp
+++ b/cpp/src/qpid/broker/BrokerAdapter.cpp
@@ -182,6 +182,7 @@ void BrokerAdapter::QueueHandlerImpl::declare(const MethodContext& context, uint
 
 	    //add default binding:
 	    broker.getExchanges().getDefault()->bind(queue, name, 0);
+            queue->bound(broker.getExchanges().getDefault()->getName(), name, arguments);
 
             //handle automatic cleanup:
 	    if (exclusive) {
@@ -212,8 +213,11 @@ void BrokerAdapter::QueueHandlerImpl::bind(const MethodContext& context, uint16_
     Exchange::shared_ptr exchange = broker.getExchanges().get(exchangeName);
     if(exchange){
         string exchangeRoutingKey = routingKey.empty() && queueName.empty() ? queue->getName() : routingKey;
-        if (exchange->bind(queue, exchangeRoutingKey, &arguments) && exchange->isDurable() && queue->isDurable()) {
-            broker.getStore().bind(*exchange, *queue, routingKey, arguments);
+        if (exchange->bind(queue, exchangeRoutingKey, &arguments)) {
+            queue->bound(exchangeName, routingKey, arguments);
+            if (exchange->isDurable() && queue->isDurable()) {
+                broker.getStore().bind(*exchange, *queue, routingKey, arguments);
+            }
         }
         if(!nowait) client.bindOk(context.getRequestId());    
     }else{
@@ -269,6 +273,7 @@ void BrokerAdapter::QueueHandlerImpl::delete_(const MethodContext& context, uint
         count = q->getMessageCount();
         q->destroy();
         broker.getQueues().destroy(queue);
+        q->unbind(broker.getExchanges(), q);
     }
 
     if(!nowait)
diff --git a/cpp/src/qpid/broker/BrokerQueue.cpp b/cpp/src/qpid/broker/BrokerQueue.cpp
index 3521e63444..58a8c85fcb 100644
--- a/cpp/src/qpid/broker/BrokerQueue.cpp
+++ b/cpp/src/qpid/broker/BrokerQueue.cpp
@@ -238,6 +238,16 @@ void Queue::destroy()
     }
 }
 
+void Queue::bound(const string& exchange, const string& key, const FieldTable& args)
+{
+    bindings.add(exchange, key, args);
+}
+
+void Queue::unbind(ExchangeRegistry& exchanges, Queue::shared_ptr shared_ref)
+{
+    bindings.unbind(exchanges, shared_ref);
+}
+
 void Queue::setPolicy(std::auto_ptr<QueuePolicy> _policy)
 {
     policy = _policy;
diff --git a/cpp/src/qpid/broker/BrokerQueue.h b/cpp/src/qpid/broker/BrokerQueue.h
index e1e69cbc60..efb31ba216 100644
--- a/cpp/src/qpid/broker/BrokerQueue.h
+++ b/cpp/src/qpid/broker/BrokerQueue.h
@@ -33,6 +33,7 @@
 #include "qpid/sys/Monitor.h"
 #include "PersistableQueue.h"
 #include "QueuePolicy.h"
+#include "QueueBindings.h"
 
 // TODO aconway 2007-02-06: Use auto_ptr and boost::ptr_vector to
 // enforce ownership of Consumers.
@@ -72,6 +73,7 @@ namespace qpid {
             mutable uint64_t persistenceId;
             framing::FieldTable settings;
             std::auto_ptr<QueuePolicy> policy;            
+            QueueBindings bindings;
 
             void pop();
             void push(Message::shared_ptr& msg);
@@ -93,6 +95,8 @@ namespace qpid {
             void create(const qpid::framing::FieldTable& settings);
             void configure(const qpid::framing::FieldTable& settings);
             void destroy();
+            void bound(const string& exchange, const string& key, const qpid::framing::FieldTable& args);
+            void unbind(ExchangeRegistry& exchanges, Queue::shared_ptr shared_ref);
             /**
              * Delivers a message to the queue. Will record it as
              * enqueued if persistent then process it.
diff --git a/cpp/src/qpid/broker/Connection.cpp b/cpp/src/qpid/broker/Connection.cpp
index 15031ff243..d809ef14d9 100644
--- a/cpp/src/qpid/broker/Connection.cpp
+++ b/cpp/src/qpid/broker/Connection.cpp
@@ -96,8 +96,10 @@ void Connection::idleIn(){}
 void Connection::closed(){
     try {
         while (!exclusiveQueues.empty()) {
-            broker.getQueues().destroy(exclusiveQueues.front()->getName());
+            Queue::shared_ptr q(exclusiveQueues.front());
+            broker.getQueues().destroy(q->getName());
             exclusiveQueues.erase(exclusiveQueues.begin());
+            q->unbind(broker.getExchanges(), q);
         }
     } catch(std::exception& e) {
         QPID_LOG(error, " Unhandled exception while closing session: " <<
diff --git a/cpp/src/qpid/broker/QueueBindings.cpp b/cpp/src/qpid/broker/QueueBindings.cpp
new file mode 100644
index 0000000000..e2fcd493db
--- /dev/null
+++ b/cpp/src/qpid/broker/QueueBindings.cpp
@@ -0,0 +1,45 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "QueueBindings.h"
+#include "ExchangeRegistry.h"
+
+using qpid::framing::FieldTable;
+using std::string;
+using namespace qpid::broker;
+
+void QueueBindings::add(const string& exchange, const string& key, const FieldTable& args)
+{
+    bindings.push_back(new Binding(exchange, key, args));
+}
+
+void QueueBindings::unbind(ExchangeRegistry& exchanges, Queue::shared_ptr queue)
+{
+    for (Bindings::iterator i = bindings.begin(); i != bindings.end(); i++) {
+        try {
+            exchanges.get(i->exchange)->unbind(queue, i->key, &(i->args));
+        } catch (ChannelException&) {
+        }
+    }
+}
+
+QueueBindings::Binding::Binding(const string& _exchange, const string& _key, const FieldTable& _args)
+    : exchange(_exchange), key(_key), args(_args)
+{}
diff --git a/cpp/src/qpid/broker/QueueBindings.h b/cpp/src/qpid/broker/QueueBindings.h
new file mode 100644
index 0000000000..b9b0f7c15c
--- /dev/null
+++ b/cpp/src/qpid/broker/QueueBindings.h
@@ -0,0 +1,55 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#ifndef _QueueBindings_
+#define _QueueBindings_
+
+#include "qpid/framing/FieldTable.h"
+#include <boost/ptr_container/ptr_list.hpp>
+#include <boost/shared_ptr.hpp>
+
+namespace qpid {
+namespace broker {
+
+class ExchangeRegistry;
+class Queue;
+class QueueBindings
+{
+    struct Binding{
+        const std::string exchange;
+        const std::string key;
+        const qpid::framing::FieldTable args;
+        Binding(const std::string& exchange, const std::string& key, const qpid::framing::FieldTable& args);
+    };
+    
+    typedef boost::ptr_list<Binding> Bindings;
+    Bindings bindings;
+    
+public:
+    void add(const std::string& exchange, const std::string& key, const qpid::framing::FieldTable& args);
+    void unbind(ExchangeRegistry& exchanges, boost::shared_ptr<Queue> queue);
+};
+
+
+}
+}
+
+
+#endif