Requests to release a message that has not been acquired should be ignored.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@628659 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 42 insertions, 0 deletions

diff --git a/qpid/python/tests_0-10/message.py b/qpid/python/tests_0-10/message.py
index 65aab2870f..a3d32bdb2d 100644
--- a/qpid/python/tests_0-10/message.py
+++ b/qpid/python/tests_0-10/message.py
@@ -763,6 +763,48 @@ class MessageTests(TestBase):
         #check all 'browsed' messages are still on the queue
         self.assertEqual(5, channel.queue_query(queue="q").message_count)
 
+    def test_release_unacquired(self):
+        channel = self.channel
+
+        #create queue
+        self.queue_declare(queue = "q", exclusive=True, auto_delete=True, durable=True)
+
+        #send message
+        channel.message_transfer(content=Content(properties={'routing_key' : "q", 'delivery_mode':2}, body = "my-message"))
+
+        #create two 'browsers'
+        channel.message_subscribe(queue = "q", destination = "a", confirm_mode = 1, acquire_mode=1)
+        channel.message_flow(unit = 1, value = 0xFFFFFFFF, destination = "a")
+        channel.message_flow(unit = 0, value = 10, destination = "a")
+        queueA = self.client.queue("a")
+
+        channel.message_subscribe(queue = "q", destination = "b", confirm_mode = 1, acquire_mode=1)
+        channel.message_flow(unit = 1, value = 0xFFFFFFFF, destination = "b")
+        channel.message_flow(unit = 0, value = 10, destination = "b")
+        queueB = self.client.queue("b")
+        
+        #have each browser release the message
+        msgA = queueA.get(timeout = 1)
+        channel.message_release([msgA.command_id, msgA.command_id])
+
+        msgB = queueB.get(timeout = 1)
+        channel.message_release([msgB.command_id, msgB.command_id])
+        
+        #cancel browsers
+        channel.message_cancel(destination = "a")
+        channel.message_cancel(destination = "b")
+        
+        #create consumer
+        channel.message_subscribe(queue = "q", destination = "c", confirm_mode = 1, acquire_mode=0)
+        channel.message_flow(unit = 1, value = 0xFFFFFFFF, destination = "c")
+        channel.message_flow(unit = 0, value = 10, destination = "c")
+        queueC = self.client.queue("c")
+        #consume the message then ack it
+        msgC = queueC.get(timeout = 1)
+        msgC.complete()
+        #ensure there are no other messages
+        self.assertEmpty(queueC)
+
     def test_no_size(self):
         self.queue_declare(queue = "q", exclusive=True, auto_delete=True)