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

1 files changed, 15 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
     }
 }