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

8 files changed, 176 insertions, 3 deletions

diff --git a/cpp/src/Makefile.am b/cpp/src/Makefile.am
index cac7761588..434313b998 100644
--- a/cpp/src/Makefile.am
+++ b/cpp/src/Makefile.am
@@ -202,6 +202,7 @@ libqpidbroker_la_SOURCES = \
   qpid/broker/MessageStoreModule.cpp \
   qpid/broker/NameGenerator.cpp \
   qpid/broker/NullMessageStore.cpp \
+  qpid/broker/QueueBindings.cpp \
   qpid/broker/QueuePolicy.cpp \
   qpid/broker/QueueRegistry.cpp \
   qpid/broker/RecoveryManagerImpl.cpp \
@@ -259,6 +260,7 @@ nobase_include_HEADERS = \
   qpid/broker/NullMessageStore.h \
   qpid/broker/Persistable.h \
   qpid/broker/Prefetch.h \
+  qpid/broker/QueueBindings.h \
   qpid/broker/QueueRegistry.h \
   qpid/broker/RecoverableExchange.h \
   qpid/broker/RecoverableMessage.h \
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
diff --git a/cpp/src/tests/QueueTest.cpp b/cpp/src/tests/QueueTest.cpp
index dd63cbaa05..7d6f5f4672 100644
--- a/cpp/src/tests/QueueTest.cpp
+++ b/cpp/src/tests/QueueTest.cpp
@@ -18,13 +18,19 @@
  * under the License.
  *
  */
+#include "qpid/Exception.h"
 #include "qpid/broker/BrokerQueue.h"
+#include "qpid/broker/Deliverable.h"
+#include "qpid/broker/ExchangeRegistry.h"
 #include "qpid/broker/QueueRegistry.h"
 #include "qpid_test_plugin.h"
 #include <iostream>
 #include "MockChannel.h"
+#include "boost/format.hpp"
 
+using namespace qpid;
 using namespace qpid::broker;
+using namespace qpid::framing;
 using namespace qpid::sys;
 
 
@@ -35,6 +41,14 @@ public:
     virtual bool deliver(Message::shared_ptr& msg);
 };
 
+class FailOnDeliver : public Deliverable
+{
+public:
+    void deliverTo(Queue::shared_ptr& queue)
+    {
+        throw Exception(boost::format("Invalid delivery to %1%") % queue->getName());
+    }
+};
 
 class QueueTest : public CppUnit::TestCase  
 {
@@ -42,8 +56,10 @@ class QueueTest : public CppUnit::TestCase
     CPPUNIT_TEST(testConsumers);
     CPPUNIT_TEST(testRegistry);
     CPPUNIT_TEST(testDequeue);
+    CPPUNIT_TEST(testBound);
     CPPUNIT_TEST_SUITE_END();
 
+
   public:
     Message::shared_ptr message(std::string exchange, std::string routingKey) {
         return Message::shared_ptr(
@@ -135,6 +151,40 @@ class QueueTest : public CppUnit::TestCase
         CPPUNIT_ASSERT_EQUAL(uint32_t(0), queue->getMessageCount());
         
     }
+
+    void testBound()
+    {
+        //test the recording of bindings, and use of those to allow a queue to be unbound
+        string key("my-key");
+        FieldTable args;
+
+        Queue::shared_ptr queue(new Queue("my-queue", true));
+        ExchangeRegistry exchanges;
+        //establish bindings from exchange->queue and notify the queue as it is bound:
+        Exchange::shared_ptr exchange1 = exchanges.declare("my-exchange-1", "direct").first;
+        exchange1->bind(queue, key, &args);
+        queue->bound(exchange1->getName(), key, args);
+
+        Exchange::shared_ptr exchange2 = exchanges.declare("my-exchange-2", "fanout").first;
+        exchange2->bind(queue, key, &args);
+        queue->bound(exchange2->getName(), key, args);
+
+        Exchange::shared_ptr exchange3 = exchanges.declare("my-exchange-3", "topic").first;
+        exchange3->bind(queue, key, &args);
+        queue->bound(exchange3->getName(), key, args);
+
+        //delete one of the exchanges:
+        exchanges.destroy(exchange2->getName());
+        exchange2.reset();
+
+        //unbind the queue from all exchanges it knows it has been bound to:
+        queue->unbind(exchanges, queue);
+
+        //ensure the remaining exchanges don't still have the queue bound to them:
+        FailOnDeliver deliverable;        
+        exchange1->route(deliverable, key, &args);
+        exchange3->route(deliverable, key, &args);
+    }
 };
 
 // Make this test suite a plugin.