Some restructuring of the client code:

 * Introduced three separate 'handlers' for the connection, channel and execution 'layers'.
 * Support for asynchronous retrieval of response or completion status.
 * Channel methods no longer included in execution layers command id count.



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

5 files changed, 18 insertions, 18 deletions

diff --git a/cpp/src/tests/FramingTest.cpp b/cpp/src/tests/FramingTest.cpp
index 98f89b59be..582c7d6e55 100644
--- a/cpp/src/tests/FramingTest.cpp
+++ b/cpp/src/tests/FramingTest.cpp
@@ -410,12 +410,14 @@ class FramingTest : public CppUnit::TestCase
         ASSERT_FRAME("BROKER: Frame[channel=1; ChannelOpenOk: ]", *i++);
         ASSERT_FRAME("CLIENT: Frame[channel=1; ExchangeDeclare: ticket=0; exchange=MyExchange; type=topic; alternateExchange=; passive=0; durable=0; autoDelete=0; arguments={}]", *i++);
         ASSERT_FRAME("CLIENT: Frame[channel=1; ExecutionFlush: ]", *i++);
-        ASSERT_FRAME("BROKER: Frame[channel=1; ExecutionComplete: cumulativeExecutionMark=2; rangedExecutionSet={}]", *i++);
+        ASSERT_FRAME("BROKER: Frame[channel=1; ExecutionComplete: cumulativeExecutionMark=1; rangedExecutionSet={}]", *i++);
         ASSERT_FRAME("CLIENT: Frame[channel=1; QueueDeclare: ticket=0; queue=MyQueue; alternateExchange=; passive=0; durable=0; exclusive=1; autoDelete=1; nowait=0; arguments={}]", *i++);
         ASSERT_FRAME("BROKER: Frame[channel=1; QueueDeclareOk: queue=MyQueue; messageCount=0; consumerCount=0]", *i++);
+        ASSERT_FRAME("CLIENT: Frame[channel=1; ExecutionFlush: ]", *i++);
+        ASSERT_FRAME("BROKER: Frame[channel=1; ExecutionComplete: cumulativeExecutionMark=2; rangedExecutionSet={}]", *i++);
         ASSERT_FRAME("CLIENT: Frame[channel=1; QueueBind: ticket=0; queue=MyQueue; exchange=MyExchange; routingKey=MyTopic; arguments={}]", *i++);
         ASSERT_FRAME("CLIENT: Frame[channel=1; ExecutionFlush: ]", *i++);
-        ASSERT_FRAME("BROKER: Frame[channel=1; ExecutionComplete: cumulativeExecutionMark=4; rangedExecutionSet={}]", *i++);
+        ASSERT_FRAME("BROKER: Frame[channel=1; ExecutionComplete: cumulativeExecutionMark=3; rangedExecutionSet={}]", *i++);
     }
  };
 
diff --git a/cpp/src/tests/Serializer.cpp b/cpp/src/tests/Serializer.cpp
index 8c0ee7b85c..d7345acf06 100644
--- a/cpp/src/tests/Serializer.cpp
+++ b/cpp/src/tests/Serializer.cpp
@@ -61,12 +61,17 @@ struct Tester {
     }
 };
 
+void execute(Serializer& s, Serializer::Task t) 
+{
+    s.execute(t);
+}
+
 BOOST_AUTO_TEST_CASE(testSingleThread) {
     // Verify that we call in the same thread by default.
     Tester tester;
     Serializer s;
     for (int i = 0; i < 100; ++i) 
-        s.execute(boost::bind(&Tester::test, &tester));
+        execute(s, boost::bind(&Tester::test, &tester));
     // All should be executed in this thread.
     BOOST_CHECK_EQUAL(0u, tester.collisions);
     BOOST_CHECK_EQUAL(100u, tester.count);
@@ -80,7 +85,7 @@ BOOST_AUTO_TEST_CASE(testSingleThreadNoImmediate) {
     Tester tester;
     Serializer s(false);
     for (int i = 0; i < 100; ++i)
-        s.execute(boost::bind(&Tester::test, &tester));
+        execute(s, boost::bind(&Tester::test, &tester));
     {
         // Wait for dispatch thread to complete.
         Mutex::ScopedLock l(tester.lock);
@@ -95,7 +100,7 @@ BOOST_AUTO_TEST_CASE(testSingleThreadNoImmediate) {
 
 struct Caller : public Runnable, public Tester {
     Caller(Serializer& s) : serializer(s) {}
-    void run() { serializer.execute(boost::bind(&Tester::test, this)); }
+    void run() { execute(serializer, boost::bind(&Tester::test, this)); }
     Serializer& serializer;
 };
 
@@ -134,7 +139,7 @@ BOOST_AUTO_TEST_CASE(testExternalDispatch) {
     serializer.reset(new Serializer(false, &notifyDispatch));
     Tester tester;
     for (int i = 0; i < 100; ++i) 
-        serializer->execute(boost::bind(&Tester::test, &tester));
+        execute(*serializer, boost::bind(&Tester::test, &tester));
     {
         // Wait for dispatch thread to complete.
         Mutex::ScopedLock l(tester.lock);
diff --git a/cpp/src/tests/client_test.cpp b/cpp/src/tests/client_test.cpp
index cefc4338eb..4903312cd7 100644
--- a/cpp/src/tests/client_test.cpp
+++ b/cpp/src/tests/client_test.cpp
@@ -41,7 +41,6 @@ using namespace qpid::client;
 using namespace qpid::sys;
 using std::string;
 
-bool verbose = false;
 
 /**
  * A simple message listener implementation that prints out the
@@ -50,9 +49,10 @@ bool verbose = false;
  */
 class SimpleListener : public virtual MessageListener{
     Monitor* monitor;
+    bool verbose;
 
 public:
-    inline SimpleListener(Monitor* _monitor) : monitor(_monitor){}
+    inline SimpleListener(Monitor* _monitor, bool debug) : monitor(_monitor), verbose(debug) {}
 
     inline virtual void received(Message& msg){
         if (verbose)
@@ -101,7 +101,7 @@ int main(int argc, char** argv)
 	//montior to use to notify the main thread when that message
 	//is received.
 	Monitor monitor;
-	SimpleListener listener(&monitor);
+	SimpleListener listener(&monitor, opts.trace);
 	string tag("MyTag");
 	channel.consume(queue, tag, &listener);
 	if (opts.trace) std::cout << "Registered consumer." << std::endl;
@@ -118,11 +118,6 @@ int main(int argc, char** argv)
 	msg.setData(data);
 	channel.publish(msg, exchange, "MyTopic");
 	if (opts.trace) std::cout << "Published message: " << data << std::endl;
-        if (opts.trace) {
-            std::cout << "Publication " 
-                      << (channel.synchWithServer(qpid::sys::TIME_SEC * 1) ? " DID " : " did NOT ") 
-                      << "complete"  << std::endl;
-        }
 
 	{
             Monitor::ScopedLock l(monitor);
diff --git a/cpp/src/tests/topic_listener.cpp b/cpp/src/tests/topic_listener.cpp
index cb6bafcd8e..cddf3cb92a 100644
--- a/cpp/src/tests/topic_listener.cpp
+++ b/cpp/src/tests/topic_listener.cpp
@@ -111,8 +111,7 @@ int main(int argc, char** argv){
             channel.bind(Exchange::STANDARD_TOPIC_EXCHANGE, control, "topic_control", bindArgs);
             //set up listener
             Listener listener(&channel, response.getName(), args.transactional);
-            string tag;
-            channel.consume(control, tag, &listener, AckMode(args.ackmode));
+            channel.consume(control, "c1", &listener, AckMode(args.ackmode));
             cout << "topic_listener: Consuming." << endl;
             channel.run();
             connection.close();
diff --git a/cpp/src/tests/topic_publisher.cpp b/cpp/src/tests/topic_publisher.cpp
index f792540c09..5800f9225d 100644
--- a/cpp/src/tests/topic_publisher.cpp
+++ b/cpp/src/tests/topic_publisher.cpp
@@ -121,8 +121,7 @@ int main(int argc, char** argv) {
 
             //set up listener
             Publisher publisher(&channel, "topic_control", args.transactional);
-            string tag("mytag");
-            channel.consume(response, tag, &publisher, AckMode(args.ackmode));
+            channel.consume(response, "mytag", &publisher, AckMode(args.ackmode));
             channel.start();
 
             int batchSize(args.batches);