Author: gsim
Date: Thu Jun 14 00:12:14 2007
New Revision: 547151
URL: http://svn.apache.org/viewvc?view=rev&rev=547151
Log:
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
Added:
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.cpp (with
props)
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.h (with props)
Modified:
incubator/qpid/trunk/qpid/cpp/src/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
Modified: incubator/qpid/trunk/qpid/cpp/src/Makefile.am
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/Makefile.am?view=diff&rev=547151&r1=547150&r2=547151
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/Makefile.am Thu Jun 14 00:12:14 2007
@@ -202,6 +202,7 @@
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 @@
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 \
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp?view=diff&rev=547151&r1=547150&r2=547151
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp Thu Jun 14
00:12:14 2007
@@ -182,6 +182,7 @@
//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 @@
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 @@
count = q->getMessageCount();
q->destroy();
broker.getQueues().destroy(queue);
+ q->unbind(broker.getExchanges(), q);
}
if(!nowait)
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp?view=diff&rev=547151&r1=547150&r2=547151
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.cpp Thu Jun 14
00:12:14 2007
@@ -238,6 +238,16 @@
}
}
+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;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h?view=diff&rev=547151&r1=547150&r2=547151
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h Thu Jun 14
00:12:14 2007
@@ -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 @@
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 @@
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.
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp?view=diff&rev=547151&r1=547150&r2=547151
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp Thu Jun 14
00:12:14 2007
@@ -96,8 +96,10 @@
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: " <<
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.cpp?view=auto&rev=547151
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.cpp Thu Jun 14
00:12:14 2007
@@ -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)
+{}
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.h?view=auto&rev=547151
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.h Thu Jun 14
00:12:14 2007
@@ -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
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/QueueBindings.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp?view=diff&rev=547151&r1=547150&r2=547151
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/QueueTest.cpp Thu Jun 14 00:12:14
2007
@@ -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 @@
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 @@
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(
@@ -134,6 +150,40 @@
CPPUNIT_ASSERT(!received);
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);
}
};
