[QPID-1909] Consumer with byte credit can get ignored if a large message "eclipses" a small one.

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Won't Fix
• Affects Version/s: 0.5
• Fix Version/s: Future
• Component/s: C++ Broker
• Assignee: Gordon Sim
• Reporter: Alan Conway
• Created: 17/Jun/09 17:55
• Updated: 02/May/16 12:51
• Resolved: 02/May/16 12:51

Description

Given: A consumer with byte credit N for a queue with messages Big size > N and Small size < N
The consumer should not be able to consume Big. Now remove Big with a different consumer.
The consumer should now be able to consume Small, but doesn't

The problem in Queue.cpp is twofold:

• if a consumer returns "false" from accept, it is assumed to be blocked and is removed from the listener set. It won't be notified of new messages
• the queue only notifies when new messages arrive. It does not notify when a blocking message is removed.

This is demonstrated by adding the following to test_credit_flow_bytes in message.py:

1. Check for bug that allowed a small message to be hidden by a larger one.
   session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "toobigtoobigtoobigtoobig"))
   session.message_transfer(message=Message(session.delivery_properties(routing_key="q"), "abcdefgh"))
   session.message_flow(unit = session.credit_unit.byte, value = msg_size, destination = "c")
   try:
   q.get(timeout = 1) # Should fail, message is too big.
   self.fail("Expected Empty exception")
   except Empty: None

1. Create another consumer to remove the large message
   session.message_subscribe(queue = "q", destination = "c2")
   session.message_flow(unit = session.credit_unit.byte, value = 0xFFFFFFFFL, destination = "c2")
   session.message_flow(unit = session.credit_unit.message, value = 1, destination = "c2")
   self.assertDataEquals(session, session.incoming("c2").get(timeout=1), "toobigtoobigtoobigtoobig")

1. Now the first consumer should be able to receive the smaller message.
   self.assertDataEquals(session, q.get(timeout = 1), "abcdefgh")