Bug 471300

+++ This bug was initially created as a clone of Bug #441700 +++

The core feature would be to have a fixed size queue (fixed in the number of
unacked messages). Once this limit was reached, new publications would result in
the oldest message being dropped.

In addition a further option would prevent the replacement of oldest by newest
message when a client was processing the messages. Publication would fail in
this state.

--- Additional comment from gsim on 2008-04-09 11:33:21 EDT ---

The objective of the second option above is to prevent 'gaps' appearing in the
sequence seen by the consuming client.

--- Additional comment from sergei.vorobiev on 2008-06-19 10:28:45 EDT ---

otherwise the ring queue is expected to have the same basic features as a
regular queue (such as acquire modes, transactional delivery, ability to
cluster, transient vs durable, etc)

--- Additional comment from gsim on 2008-09-30 10:44:16 EDT ---

There are two 'policy types' through which these features are delivered, these and the desired limits, are passed as name-value pairs in the arguments parameter to the queue.declare command used to create the queue.

qpid.policy_type=ring will cause the queue to delete the oldest message when an enqueue would otherwise breach the configured queue limit.

qpid.policy_type=ring_strict does the same, but if the oldest message has been delivered to a client but not yet acknowledged, it will not be removed to make way for new messages and the attempted enqueue will result in a failure.

The limit is set through either qpid.max_size (the total size of all messages on the queue) or qpid.max_count (the number of messages on the queue).

The functionality is covered by the QueuePolicyTests suite in Qpid, particularly the testRingPolicy and testStrictRingPolicy test cases. 

Test strategy: 

Case 1:
1.1. configure a queue with qpid.policy_type=ring, qpid.count=5
1.2. publish 10 distinct messages
1.3. read the messages from the queue and verify that only the last five were delivered

Case 2:

As for case 1, but using qpid.policy_type=ring_strict in step 1

Case 3:

1. configure a queue with qpid.policy_type=ring_strict, qpid.count=5
2. publish 5 messages
3. subscribe to queue and receive but do not ack all 5 messages
4. attempt to publish a further message which should fail

Case 4:

1. configure a queue with qpid.policy_type=ring_strict, qpid.count=5
2. publish 5 messages
3. subscribe to queue and receive but do not ack all 5 messages
4. attempt to publish a further message which should succeed
5. ack all messages