Merged in channel.flow implementation and interoperability tests.

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

4 files changed, 21 insertions, 6 deletions

diff --git a/cpp/src/qpid/broker/BrokerAdapter.cpp b/cpp/src/qpid/broker/BrokerAdapter.cpp
index 3c742b8d2d..592995f10f 100644
--- a/cpp/src/qpid/broker/BrokerAdapter.cpp
+++ b/cpp/src/qpid/broker/BrokerAdapter.cpp
@@ -100,7 +100,11 @@ void BrokerAdapter::ChannelHandlerImpl::open(
         std::string()/* ID */, context.getRequestId());
 } 
         
-void BrokerAdapter::ChannelHandlerImpl::flow(const MethodContext&, bool /*active*/){}         
+void BrokerAdapter::ChannelHandlerImpl::flow(const MethodContext& context, bool active){
+    channel.flow(active);
+    client.flowOk(active, context.getRequestId());
+}         
+
 void BrokerAdapter::ChannelHandlerImpl::flowOk(const MethodContext&, bool /*active*/){} 
         
 void BrokerAdapter::ChannelHandlerImpl::close(
diff --git a/cpp/src/qpid/broker/BrokerChannel.cpp b/cpp/src/qpid/broker/BrokerChannel.cpp
index 0c06350c02..e256566d35 100644
--- a/cpp/src/qpid/broker/BrokerChannel.cpp
+++ b/cpp/src/qpid/broker/BrokerChannel.cpp
@@ -66,6 +66,7 @@ Channel::Channel(
     store(_store),
     messageBuilder(this, _store, _stagingThreshold),
     opened(id == 0),//channel 0 is automatically open, other must be explicitly opened
+    flowActive(true),
     adapter(new BrokerAdapter(*this, con, con.broker))
 {
     outstanding.reset();
@@ -221,7 +222,7 @@ Channel::ConsumerImpl::ConsumerImpl(Channel* _parent, const string& _tag,
 
 bool Channel::ConsumerImpl::deliver(Message::shared_ptr& msg){
     if(!connection || connection != msg->getPublisher()){//check for no_local
-        if(ackExpected && !parent->checkPrefetch(msg)){
+        if(!parent->flowActive || (ackExpected && !parent->checkPrefetch(msg))){
             blocked = true;
         }else{
             blocked = false;
@@ -396,3 +397,14 @@ void Channel::handleMethodInContext(
         connection.close(541/*internal error*/, e.what(), method->amqpClassId(), method->amqpMethodId());
     }
 }
+
+void Channel::flow(bool active)
+{
+    Mutex::ScopedLock locker(deliveryLock);
+    bool requestDelivery(!flowActive && active);
+    flowActive = active;
+    if (requestDelivery) {
+        //there may be messages that can be now be delivered
+        std::for_each(consumers.begin(), consumers.end(), boost::bind(&ConsumerImpl::requestDispatch, _1));
+    }
+}
diff --git a/cpp/src/qpid/broker/BrokerChannel.h b/cpp/src/qpid/broker/BrokerChannel.h
index a2f17f85f4..1fbfc2063e 100644
--- a/cpp/src/qpid/broker/BrokerChannel.h
+++ b/cpp/src/qpid/broker/BrokerChannel.h
@@ -97,6 +97,7 @@ class Channel : public framing::ChannelAdapter,
     MessageStore* const store;
     MessageBuilder messageBuilder;//builder for in-progress message
     bool opened;
+    bool flowActive;
     boost::scoped_ptr<BrokerAdapter> adapter;
 
 	// completion handler for MessageBuilder
@@ -147,6 +148,7 @@ class Channel : public framing::ChannelAdapter,
     void ack(uint64_t deliveryTag, bool multiple);
     void ack(uint64_t deliveryTag, uint64_t endTag);
     void recover(bool requeue);
+    void flow(bool active);
     void deliver(Message::shared_ptr& msg, const string& consumerTag, uint64_t deliveryTag);            
     void handlePublish(Message* msg);
     void handleHeader(boost::shared_ptr<framing::AMQHeaderBody>);
diff --git a/cpp/src/qpid/broker/BrokerQueue.cpp b/cpp/src/qpid/broker/BrokerQueue.cpp
index ee1a913a96..3521e63444 100644
--- a/cpp/src/qpid/broker/BrokerQueue.cpp
+++ b/cpp/src/qpid/broker/BrokerQueue.cpp
@@ -78,10 +78,7 @@ bool Queue::dispatch(Message::shared_ptr& msg){
     if(consumers.empty()){
         return false;
     }else if(exclusive){
-        if(!exclusive->deliver(msg)){
-            QPID_LOG(warning, "Dropping undeliverable message from queue with exclusive consumer.");
-        }
-        return true;
+        return exclusive->deliver(msg);
     }else{
         //deliver to next consumer
         next = next % consumers.size();