diff options
-rw-r--r-- | qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp | 3 | ||||
-rw-r--r-- | qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp | 19 |
2 files changed, 18 insertions, 4 deletions
diff --git a/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp b/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp index 665bf2def4..9605cacac1 100644 --- a/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp +++ b/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp @@ -145,6 +145,7 @@ void Outgoing::detached() bool Outgoing::deliver(const QueueCursor& cursor, const qpid::broker::Message& msg) { Record& r = deliveries[current++]; + if (current >= deliveries.capacity()) current = 0; r.cursor = cursor; r.msg = msg; pn_delivery(link, r.tag); @@ -161,7 +162,7 @@ void Outgoing::notify() bool Outgoing::accept(const qpid::broker::Message&) { - return canDeliver(); + return true; } void Outgoing::setSubjectFilter(const std::string& f) diff --git a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp index 8ad63e325f..173fcba552 100644 --- a/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp +++ b/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp @@ -188,6 +188,7 @@ bool ConnectionContext::fetch(boost::shared_ptr<SessionContext> ssn, boost::shar qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock); if (lnk->capacity) { pn_link_flow(lnk->receiver, 1);//TODO: is this the right approach? + wakeupDriver(); } return true; } else { @@ -195,12 +196,24 @@ bool ConnectionContext::fetch(boost::shared_ptr<SessionContext> ssn, boost::shar qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock); pn_link_drain(lnk->receiver, 0); wakeupDriver(); - while (pn_link_credit((pn_link_t*) lnk->receiver) - pn_link_queued((pn_link_t*) lnk->receiver)) { - QPID_LOG(notice, "Waiting for credit to be drained: " << (pn_link_credit((pn_link_t*) lnk->receiver) - pn_link_queued((pn_link_t*) lnk->receiver))); + while (pn_link_credit(lnk->receiver) && !pn_link_queued(lnk->receiver)) { + QPID_LOG(debug, "Waiting for message or for credit to be drained: credit=" << pn_link_credit(lnk->receiver) << ", queued=" << pn_link_queued(lnk->receiver)); wait(); } + if (lnk->capacity && pn_link_queued(lnk->receiver) == 0) { + pn_link_flow(lnk->receiver, lnk->capacity); + } + } + if (get(ssn, lnk, message, qpid::messaging::Duration::IMMEDIATE)) { + qpid::sys::ScopedLock<qpid::sys::Monitor> l(lock); + if (lnk->capacity) { + pn_link_flow(lnk->receiver, 1); + wakeupDriver(); + } + return true; + } else { + return false; } - return get(ssn, lnk, message, qpid::messaging::Duration::IMMEDIATE); } } |