qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From g...@apache.org
Subject svn commit: r732297 - in /qpid/trunk/qpid/cpp/src: qpid/broker/DirectExchange.cpp qpid/sys/CopyOnWriteArray.h tests/ClientSessionTest.cpp
Date Wed, 07 Jan 2009 10:46:54 GMT
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()
 
 



Mime
View raw message