Author: gsim
Date: Wed Jan 7 02:46:53 2009
New Revision: 732297
URL: http://svn.apache.org/viewvc?rev=732297&view=rev
Log:
QPID-1560: add support for a qpid.exclusive-binding option on direct exchange
that causes the binding specified to be the only one for the given key. I.e. if
there is already a binding at this exchange with this key it will be atomically
updated to bind the new queue.
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/DirectExchange.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h
qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/DirectExchange.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/DirectExchange.cpp?rev=732297&r1=732296&r2=732297&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/DirectExchange.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/DirectExchange.cpp Wed Jan 7 02:46:53
2009
@@ -33,6 +33,7 @@
const std::string qpidFedOp("qpid.fed.op");
const std::string qpidFedTags("qpid.fed.tags");
const std::string qpidFedOrigin("qpid.fed.origin");
+const std::string qpidExclusiveBinding("qpid.exclusive-binding");
const std::string fedOpBind("B");
const std::string fedOpUnbind("U");
@@ -56,15 +57,25 @@
bool DirectExchange::bind(Queue::shared_ptr queue, const string& routingKey,
const FieldTable* args)
{
- string fedOp(args ? args->getAsString(qpidFedOp) : fedOpBind);
- string fedTags(args ? args->getAsString(qpidFedTags) : "");
- string fedOrigin(args ? args->getAsString(qpidFedOrigin) : "");
+ string fedOp(fedOpBind);
+ string fedTags;
+ string fedOrigin;
+ bool exclusiveBinding = false;
+ if (args) {
+ fedOp = args->getAsString(qpidFedOp);
+ fedTags = args->getAsString(qpidFedTags);
+ fedOrigin = args->getAsString(qpidFedOrigin);
+ exclusiveBinding = args->get(qpidExclusiveBinding);
+ }
+
bool propagate = false;
if (args == 0 || fedOp.empty() || fedOp == fedOpBind) {
Mutex::ScopedLock l(lock);
Binding::shared_ptr b(new Binding(routingKey, queue, this,
FieldTable(), fedOrigin));
BoundKey& bk = bindings[routingKey];
+ if (exclusiveBinding) bk.queues.clear();
+
if (bk.queues.add_unless(b, MatchQueue(queue))) {
propagate = bk.fedBinding.addOrigin(fedOrigin);
if (mgmtExchange != 0) {
Modified: qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h?rev=732297&r1=732296&r2=732297&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/CopyOnWriteArray.h Wed Jan 7 02:46:53 2009
@@ -64,6 +64,18 @@
}
}
+ bool clear()
+ {
+ Mutex::ScopedLock l(lock);
+ if (array && !array->empty()) {
+ ArrayPtr copy;
+ array = copy;
+ return true;
+ } else {
+ return false;
+ }
+ }
+
template <class F>
bool add_unless(T& t, F f)
{
Modified: qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp?rev=732297&r1=732296&r2=732297&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp Wed Jan 7 02:46:53 2009
@@ -447,6 +447,28 @@
}
+QPID_AUTO_TEST_CASE(testExclusiveBinding) {
+ FieldTable options;
+ options.setString("qpid.exclusive-binding", "anything");
+ ClientSessionFixture fix;
+ fix.session.queueDeclare(arg::queue="queue-1", arg::exclusive=true,
arg::autoDelete=true);
+ fix.session.queueDeclare(arg::queue="queue-2", arg::exclusive=true,
arg::autoDelete=true);
+ fix.session.exchangeBind(arg::exchange="amq.direct", arg::queue="queue-1",
arg::bindingKey="my-key", arg::arguments=options);
+ fix.session.messageTransfer(arg::destination="amq.direct",
arg::content=Message("message1", "my-key"));
+ fix.session.exchangeBind(arg::exchange="amq.direct", arg::queue="queue-2",
arg::bindingKey="my-key", arg::arguments=options);
+ fix.session.messageTransfer(arg::destination="amq.direct",
arg::content=Message("message2", "my-key"));
+
+ Message got;
+ BOOST_CHECK(fix.subs.get(got, "queue-1"));
+ BOOST_CHECK_EQUAL("message1", got.getData());
+ BOOST_CHECK(!fix.subs.get(got, "queue-1"));
+
+ BOOST_CHECK(fix.subs.get(got, "queue-2"));
+ BOOST_CHECK_EQUAL("message2", got.getData());
+ BOOST_CHECK(!fix.subs.get(got, "queue-2"));
+}
+
+
QPID_AUTO_TEST_SUITE_END()
