Return-Path: X-Original-To: apmail-qpid-commits-archive@www.apache.org Delivered-To: apmail-qpid-commits-archive@www.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id 30EC2E89C for ; Wed, 28 Nov 2012 14:14:04 +0000 (UTC) Received: (qmail 36868 invoked by uid 500); 28 Nov 2012 14:14:04 -0000 Delivered-To: apmail-qpid-commits-archive@qpid.apache.org Received: (qmail 36815 invoked by uid 500); 28 Nov 2012 14:14:03 -0000 Mailing-List: contact commits-help@qpid.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@qpid.apache.org Delivered-To: mailing list commits@qpid.apache.org Received: (qmail 36807 invoked by uid 99); 28 Nov 2012 14:14:03 -0000 Received: from nike.apache.org (HELO nike.apache.org) (192.87.106.230) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 28 Nov 2012 14:14:03 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=5.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [140.211.11.4] (HELO eris.apache.org) (140.211.11.4) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 28 Nov 2012 14:14:01 +0000 Received: from eris.apache.org (localhost [127.0.0.1]) by eris.apache.org (Postfix) with ESMTP id 2DD912388ABA; Wed, 28 Nov 2012 14:13:40 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r1414708 - in /qpid/branches/0.20/qpid/cpp/src/qpid: broker/amqp/Outgoing.cpp messaging/amqp/ConnectionContext.cpp Date: Wed, 28 Nov 2012 14:13:39 -0000 To: commits@qpid.apache.org From: gsim@apache.org X-Mailer: svnmailer-1.0.8-patched Message-Id: <20121128141340.2DD912388ABA@eris.apache.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: gsim Date: Wed Nov 28 14:13:38 2012 New Revision: 1414708 URL: http://svn.apache.org/viewvc?rev=1414708&view=rev Log: QPID-4460: Replenish credit to cover specified prefetch if it is drained Modified: qpid/branches/0.20/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp qpid/branches/0.20/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp Modified: qpid/branches/0.20/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp?rev=1414708&r1=1414707&r2=1414708&view=diff ============================================================================== --- qpid/branches/0.20/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp (original) +++ qpid/branches/0.20/qpid/cpp/src/qpid/broker/amqp/Outgoing.cpp Wed Nov 28 14:13:38 2012 @@ -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) Modified: qpid/branches/0.20/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp URL: http://svn.apache.org/viewvc/qpid/branches/0.20/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp?rev=1414708&r1=1414707&r2=1414708&view=diff ============================================================================== --- qpid/branches/0.20/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp (original) +++ qpid/branches/0.20/qpid/cpp/src/qpid/messaging/amqp/ConnectionContext.cpp Wed Nov 28 14:13:38 2012 @@ -188,6 +188,7 @@ bool ConnectionContext::fetch(boost::sha qpid::sys::ScopedLock 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::sha qpid::sys::ScopedLock 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 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); } } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org For additional commands, e-mail: commits-help@qpid.apache.org