Patch from Gordon Sim to fix issues with hasOutput implementation.

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

1 files changed, 23 insertions, 2 deletions

diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index 3b447e97f2..acab1e06f1 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -212,9 +212,30 @@ bool Queue::getNextMessage(QueuedMessage& m, Consumer& c)
     }
 }
 
-bool Queue::empty() const {
+bool Queue::checkForMessages(Consumer& c)
+{
     Mutex::ScopedLock locker(messageLock);
-    return messages.empty();
+    if (messages.empty()) {
+        //no message available, register consumer for notification
+        //when this changes
+        addListener(c);
+        return false;
+    } else {
+        QueuedMessage msg = messages.front();
+        if (store && !msg.payload->isEnqueueComplete()) {
+            //though a message is on the queue, it has not yet been
+            //enqueued and so is not available for consumption yet,
+            //register consumer for notification when this changes
+            addListener(c);
+            return false;            
+        } else {
+            //check that consumer has sufficient credit for the
+            //message (if it does not, no need to register it for
+            //notification as the consumer itself will handle the
+            //credit allocation required to change this condition).
+            return c.accept(msg.payload);
+        }
+    }
 }
 
 bool Queue::consumeNextMessage(QueuedMessage& m, Consumer& c)