Additions to the client API:

 - SubscriptionManager::get(queue) to get a single message from a queue.
 - Set FlowControl per-subscription.


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

6 files changed, 69 insertions, 23 deletions

diff --git a/cpp/src/tests/ClientSessionTest.cpp b/cpp/src/tests/ClientSessionTest.cpp
index 83c3317094..0475350d6a 100644
--- a/cpp/src/tests/ClientSessionTest.cpp
+++ b/cpp/src/tests/ClientSessionTest.cpp
@@ -203,7 +203,7 @@ QPID_AUTO_TEST_CASE(testSendToSelf) {
     ClientSessionFixture fix;
     SimpleListener mylistener;
     fix.session.queueDeclare(queue="myq", exclusive=true, autoDelete=true);
-    fix.subs.subscribe(mylistener, "myq", "myq");
+    fix.subs.subscribe(mylistener, "myq");
     sys::Thread runner(fix.subs);//start dispatcher thread
     string data("msg");
     Message msg(data, "myq");
@@ -222,5 +222,30 @@ QPID_AUTO_TEST_CASE(testSendToSelf) {
     }
 }
 
+QPID_AUTO_TEST_CASE(testLocalQueue) {
+    ClientSessionFixture fix;
+    fix.session.queueDeclare(queue="lq", exclusive=true, autoDelete=true);
+    LocalQueue lq;
+    fix.subs.subscribe(lq, "lq", FlowControl(2, FlowControl::UNLIMITED, false));
+    fix.session.messageTransfer(content=Message("foo0", "lq"));
+    fix.session.messageTransfer(content=Message("foo1", "lq"));
+    fix.session.messageTransfer(content=Message("foo2", "lq"));
+    BOOST_CHECK_EQUAL("foo0", lq.pop().getData());
+    BOOST_CHECK_EQUAL("foo1", lq.pop().getData());
+    BOOST_CHECK(lq.empty());    // Credit exhausted.
+    fix.subs.setFlowControl("lq", FlowControl::unlimited());
+    BOOST_CHECK_EQUAL("foo2", lq.pop().getData());    
+}
+
+QPID_AUTO_TEST_CASE(testGet) {
+    ClientSessionFixture fix;
+    fix.session.queueDeclare(queue="getq", exclusive=true, autoDelete=true);
+    fix.session.messageTransfer(content=Message("foo0", "getq"));
+    fix.session.messageTransfer(content=Message("foo1", "getq"));
+    BOOST_CHECK_EQUAL("foo0", fix.subs.get("getq").getData());
+    BOOST_CHECK_EQUAL("foo1", fix.subs.get("getq").getData());
+}
+
 QPID_AUTO_TEST_SUITE_END()
 
+
diff --git a/cpp/src/tests/ais_check b/cpp/src/tests/ais_check
index 344f15a94e..f16480b7f0 100755
--- a/cpp/src/tests/ais_check
+++ b/cpp/src/tests/ais_check
@@ -1,7 +1,6 @@
 #!/bin/sh
-# Check for requirements, run AIS tests if found.
-#
 
+# Check AIS requirements tests if found.
 id -nG | grep '\<ais\>' >/dev/null || \
     NOGROUP="You are not a member of the ais group."
 ps -u root | grep aisexec >/dev/null || \
@@ -24,4 +23,11 @@ EOF
     exit 0;			# A warning, not a failure.
 fi
 
-echo `dirname $0`/ais_run | newgrp ais
+# Run the tests
+srcdir=`dirname $0`
+$srcdir/start_cluster 4
+./ais_test
+ret=$?
+$srcdir/stop_cluster 
+exit $ret
+
diff --git a/cpp/src/tests/ais_run b/cpp/src/tests/ais_run
deleted file mode 100755
index 0f45edc39c..0000000000
--- a/cpp/src/tests/ais_run
+++ /dev/null
@@ -1,15 +0,0 @@
-#!/bin/sh
-#
-# Run AIS tests, assumes that ais_check has passed and we are
-# running with the ais group ID.
-#
-
-# FIXME aconway 2008-01-30: we should valgrind the cluster brokers.
-
-srcdir=`dirname $0`
-$srcdir/start_cluster 4
-./ais_test
-ret=$?
-$srcdir/stop_cluster 
-exit $ret
-
diff --git a/cpp/src/tests/cluster.mk b/cpp/src/tests/cluster.mk
index d373a7d6ab..ba49a6774d 100644
--- a/cpp/src/tests/cluster.mk
+++ b/cpp/src/tests/cluster.mk
@@ -11,7 +11,7 @@ lib_cluster = $(abs_builddir)/../libqpidcluster.la
 
 # ais_check checks conditions for AIS tests and runs if ok.
 TESTS+=ais_check
-EXTRA_DIST+=ais_check ais_run start_cluster stop_cluster
+EXTRA_DIST+=ais_check start_cluster stop_cluster
 
 check_PROGRAMS+=ais_test
 ais_test_SOURCES=ais_test.cpp Cpg.cpp 
diff --git a/cpp/src/tests/cluster_client.cpp b/cpp/src/tests/cluster_client.cpp
index f6b3a80c97..efb6e04aa8 100644
--- a/cpp/src/tests/cluster_client.cpp
+++ b/cpp/src/tests/cluster_client.cpp
@@ -62,14 +62,14 @@ QPID_AUTO_TEST_CASE(testWiringReplication) {
     ClusterConnections cluster;
     BOOST_REQUIRE(cluster.size() > 1);
 
-    Session broker0 = cluster[0]->newSession(ASYNC);
+    Session broker0 = cluster[0]->newSession();
     broker0.exchangeDeclare(exchange="ex");
     broker0.queueDeclare(queue="q");
     broker0.queueBind(exchange="ex", queue="q", routingKey="key");
     broker0.close();
     
     for (size_t i = 1; i < cluster.size(); ++i) {
-        Session s = cluster[i]->newSession(ASYNC);
+        Session s = cluster[i]->newSession();
         s.messageTransfer(content=TransferContent("data", "key", "ex"));
         s.messageSubscribe(queue="q", destination="q");
         s.messageFlow(destination="q", unit=0, value=1);//messages
@@ -81,4 +81,28 @@ QPID_AUTO_TEST_CASE(testWiringReplication) {
     }    
 }
 
+QPID_AUTO_TEST_CASE(testMessageReplication) {
+    // Enqueue on one broker, dequeue on another.
+    ClusterConnections cluster;
+    BOOST_REQUIRE(cluster.size() > 1);
+
+    Session broker0 = cluster[0]->newSession();
+    broker0.queueDeclare(queue="q");
+    broker0.messageTransfer(content=TransferContent("data", "q"));
+    broker0.close();
+    
+    Session broker1 = cluster[1]->newSession();
+    broker1.
+        s.messageSubscribe(queue="q", destination="q");
+        s.messageFlow(destination="q", unit=0, value=1);//messages
+        FrameSet::shared_ptr msg = s.get();
+        BOOST_CHECK(msg->isA<MessageTransferBody>());
+        BOOST_CHECK_EQUAL(string("data"), msg->getContent());
+        s.getExecution().completed(msg->getId(), true, true);
+        cluster[i]->close();
+    }    
+}
+
+// TODO aconway 2008-06-25: dequeue replication, exactly once delivery, failover.
+
 QPID_AUTO_TEST_SUITE_END()
diff --git a/cpp/src/tests/start_cluster b/cpp/src/tests/start_cluster
index 876913bc2d..46ecbad9c5 100755
--- a/cpp/src/tests/start_cluster
+++ b/cpp/src/tests/start_cluster
@@ -3,6 +3,12 @@
 # Print the cluster's URL.
 #
 
+# Execute command with the ais group set.
+with_ais_group() {
+    id -nG | grep '\<ais\>' >/dev/null || { echo "You are not a member of the ais group."; exit 1; }
+    echo $* | newgrp ais
+}
+
 test -f cluster.ports && { echo "cluster.ports file already exists" ; exit 1; }
 test -z "$*" && { echo "Usage: $0 cluster-size [options]"; exit 1; }
 
@@ -13,7 +19,7 @@ CLUSTER=`whoami`		# Cluster name=user name, avoid clashes.
 OPTS="--load-module ../.libs/libqpidcluster.so -dp0 --log-output=cluster$i.log --cluster-name=$CLUSTER --no-data-dir --auth=no $*"
 
 for (( i=0; i<SIZE; ++i )); do
-    PORT=`../qpidd $OPTS`  || exit 1
+    PORT=`with_ais_group ../qpidd $OPTS`  || exit 1
     echo $PORT >> cluster.ports
 done