diff options
Diffstat (limited to 'cpp/src/qpid')
| -rw-r--r-- | cpp/src/qpid/client/Dispatcher.cpp | 26 | ||||
| -rw-r--r-- | cpp/src/qpid/client/SubscriptionManager.cpp | 5 | ||||
| -rw-r--r-- | cpp/src/qpid/client/SubscriptionManager.h | 4 |
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. |
