Altered topic test to use the new session api.

Exposed start() through the subscription manager in addition to run(). 



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

3 files changed, 24 insertions, 11 deletions

diff --git a/cpp/src/qpid/client/Dispatcher.cpp b/cpp/src/qpid/client/Dispatcher.cpp
index 37fada45fb..7ffcd676a3 100644
--- a/cpp/src/qpid/client/Dispatcher.cpp
+++ b/cpp/src/qpid/client/Dispatcher.cpp
@@ -69,18 +69,22 @@ void Dispatcher::run()
     boost::state_saver<bool>  reset(running); // Reset to false on exit.
     running = true;
     queue->open();
-    while (!queue->isClosed()) {
-        Mutex::ScopedUnlock u(lock);
-        FrameSet::shared_ptr content = queue->pop();
-        if (content->isA<MessageTransferBody>()) {
-            Message msg(*content, session);
-            Subscriber::shared_ptr listener = find(msg.getDestination());
-            assert(listener);
-            listener->received(msg);
-        } else {
-            assert (handler.get());
-            handler->handle(*content);
+    try {
+        while (!queue->isClosed()) {
+            Mutex::ScopedUnlock u(lock);
+            FrameSet::shared_ptr content = queue->pop();
+            if (content->isA<MessageTransferBody>()) {
+                Message msg(*content, session);
+                Subscriber::shared_ptr listener = find(msg.getDestination());
+                assert(listener);
+                listener->received(msg);
+            } else {
+                assert (handler.get());
+                handler->handle(*content);
+            }
         }
+    } catch (const ClosedException&) {
+        //ignore it and return
     }
 }
 
diff --git a/cpp/src/qpid/client/SubscriptionManager.cpp b/cpp/src/qpid/client/SubscriptionManager.cpp
index a758dc1341..438a73ec6f 100644
--- a/cpp/src/qpid/client/SubscriptionManager.cpp
+++ b/cpp/src/qpid/client/SubscriptionManager.cpp
@@ -102,6 +102,11 @@ void SubscriptionManager::stop()
     dispatcher.stop();
 }
 
+void SubscriptionManager::start()
+{
+    dispatcher.start();
+}
+
 }} // namespace qpid::client
 
 #endif
diff --git a/cpp/src/qpid/client/SubscriptionManager.h b/cpp/src/qpid/client/SubscriptionManager.h
index ad5ddfa000..fb1a726bff 100644
--- a/cpp/src/qpid/client/SubscriptionManager.h
+++ b/cpp/src/qpid/client/SubscriptionManager.h
@@ -82,9 +82,13 @@ public:
      */
     void run(bool autoStop=true);
 
+    /** Deliver messages in another thread. */
+    void start();
+
     /** Cause run() to return */
     void stop();
 
+
     static const uint32_t UNLIMITED=0xFFFFFFFF;
 
     /** Set the flow control for destination tag.