From 70a3cdf33b3e38ee26ee2840a55f83ebd26589b4 Mon Sep 17 00:00:00 2001 From: Gordon Sim Date: Wed, 6 Jun 2007 16:39:03 +0000 Subject: 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 --- cpp/src/qpid/broker/BrokerChannel.cpp | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) (limited to 'cpp/src/qpid/broker/BrokerChannel.cpp') 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)); + } +} -- cgit v1.2.1