Moved BasicMessage::WaitableDestination to IncomingMessage::WaitableDestination so it can be shared by Basic and Message implementations.

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

1 files changed, 38 insertions, 0 deletions

diff --git a/qpid/cpp/src/client/IncomingMessage.cpp b/qpid/cpp/src/client/IncomingMessage.cpp
index 05c4bc2378..eb5f2b6fae 100644
--- a/qpid/cpp/src/client/IncomingMessage.cpp
+++ b/qpid/cpp/src/client/IncomingMessage.cpp
@@ -32,6 +32,44 @@ using sys::Mutex;
 
 IncomingMessage::Destination::~Destination() {}
 
+    
+IncomingMessage::WaitableDestination::WaitableDestination()
+    : shutdownFlag(false) {}
+
+void IncomingMessage::WaitableDestination::message(const Message& msg) {
+    Mutex::ScopedLock l(monitor);
+    queue.push(msg);
+    monitor.notify();   
+}
+        
+void IncomingMessage::WaitableDestination::empty() {
+    Mutex::ScopedLock l(monitor);
+    queue.push(Empty());
+    monitor.notify();
+}
+    
+bool IncomingMessage::WaitableDestination::wait(Message& msgOut) {
+    Mutex::ScopedLock l(monitor);
+    while (queue.empty() && !shutdownFlag)
+        monitor.wait();
+    if (shutdownFlag)
+        return false;
+    Message* msg = boost::get<Message>(&queue.front());
+    bool success = msg;
+    if (success) 
+        msgOut=*msg;
+    queue.pop();
+    if (!queue.empty())
+        monitor.notify();   // Wake another waiter.
+    return success;
+}
+
+void IncomingMessage::WaitableDestination::shutdown() {
+    Mutex::ScopedLock l(monitor);
+    shutdownFlag = true;
+    monitor.notifyAll();
+}
+
 void IncomingMessage::openReference(const std::string& name) {
     Mutex::ScopedLock l(lock);
     if (references.find(name) != references.end())