Author: gsim
Date: Wed Feb 25 15:46:04 2009
New Revision: 747833
URL: http://svn.apache.org/viewvc?rev=747833&view=rev
Log:
Fix for bug in credit reallocation, along with test from [email protected]
that reproduces it.
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
qpid/trunk/qpid/python/tests_0-10/message.py
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp?rev=747833&r1=747832&r2=747833&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp Wed Feb 25 15:46:04
2009
@@ -516,14 +516,16 @@
void SemanticState::ConsumerImpl::addByteCredit(uint32_t value)
{
if (byteCredit != 0xFFFFFFFF) {
- byteCredit += value;
+ if (value == 0xFFFFFFFF) byteCredit = value;
+ else byteCredit += value;
}
}
void SemanticState::ConsumerImpl::addMessageCredit(uint32_t value)
{
if (msgCredit != 0xFFFFFFFF) {
- msgCredit += value;
+ if (value == 0xFFFFFFFF) msgCredit = value;
+ else msgCredit += value;
}
}
Modified: qpid/trunk/qpid/python/tests_0-10/message.py
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/python/tests_0-10/message.py?rev=747833&r1=747832&r2=747833&view=diff
==============================================================================
--- qpid/trunk/qpid/python/tests_0-10/message.py (original)
+++ qpid/trunk/qpid/python/tests_0-10/message.py Wed Feb 25 15:46:04 2009
@@ -462,6 +462,42 @@
self.assertDataEquals(session, q.get(timeout = 1), "abcdefgh")
self.assertEmpty(q)
+ def test_window_flush_ack_flow(self):
+ """
+ Test basic window based flow control with unit = bytes
+ """
+ #declare an exclusive queue
+ ssn = self.session
+ ssn.queue_declare(queue = "q", exclusive=True, auto_delete=True)
+ #create consumer
+ ssn.message_subscribe(queue = "q", destination = "c",
+ accept_mode=ssn.accept_mode.explicit)
+ ssn.message_set_flow_mode(flow_mode = ssn.flow_mode.window,
destination = "c")
+
+ #send message A
+
ssn.message_transfer(message=Message(ssn.delivery_properties(routing_key="q"),
"A"))
+
+ for unit in ssn.credit_unit.values():
+ ssn.message_flow("c", unit, 0xFFFFFFFFL)
+
+ q = ssn.incoming("c")
+ msgA = q.get(timeout=10)
+
+ ssn.message_flush(destination="c")
+
+ # XXX
+ ssn.receiver._completed.add(msgA.id)
+ ssn.channel.session_completed(ssn.receiver._completed)
+ ssn.message_accept(RangedSet(msgA.id))
+
+ for unit in ssn.credit_unit.values():
+ ssn.message_flow("c", unit, 0xFFFFFFFFL)
+
+ #send message B
+
ssn.message_transfer(message=Message(ssn.delivery_properties(routing_key="q"),
"B"))
+
+ msgB = q.get(timeout=10)
+
def test_subscribe_not_acquired(self):
"""
Test the not-acquired modes works as expected for a simple case
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
