qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From g...@apache.org
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 GMT
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<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::sha
             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);
     }
 }
 



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message