QPID-1660: If selected consumer can't take a message, ensure others are notified of message availability.

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

1 files changed, 12 insertions, 2 deletions

diff --git a/cpp/src/qpid/broker/SemanticState.cpp b/cpp/src/qpid/broker/SemanticState.cpp
index 13a8c649d2..4f751e43b7 100644
--- a/cpp/src/qpid/broker/SemanticState.cpp
+++ b/cpp/src/qpid/broker/SemanticState.cpp
@@ -527,9 +527,19 @@ void SemanticState::ConsumerImpl::addMessageCredit(uint32_t value)
     }
 }
 
+bool SemanticState::ConsumerImpl::haveCredit()
+{
+    if (msgCredit) {
+        return true;
+    } else {
+        blocked = true;
+        return false;
+    }
+}
+
 void SemanticState::ConsumerImpl::flush()
 {
-    while(queue->dispatch(shared_from_this()))
+    while(haveCredit() && queue->dispatch(shared_from_this()))
         ;
     stop();
 }
@@ -587,7 +597,7 @@ bool SemanticState::ConsumerImpl::hasOutput() {
 
 bool SemanticState::ConsumerImpl::doOutput()
 {
-    return queue->dispatch(shared_from_this());
+    return haveCredit() && queue->dispatch(shared_from_this());
 }
 
 void SemanticState::ConsumerImpl::enableNotify()