Fix to auto-rollback which didn't do what it was supposed to (i.e. rollback a session ended with an open transaction)

Fix to rollback to stop flow for subscriptions before the rollback, and restart after. This prevents any out of order delivery as rollback requeues the messages and will be redundant once QPID-686 is fixed.



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

1 files changed, 24 insertions, 3 deletions

diff --git a/python/tests_0-10/tx.py b/python/tests_0-10/tx.py
index 6a682e9ae7..3fd1065af3 100644
--- a/python/tests_0-10/tx.py
+++ b/python/tests_0-10/tx.py
@@ -77,8 +77,9 @@ class TxTests(TestBase):
         """
         Test that a channel closed with an open transaction is effectively rolled back
         """
-        channel = self.channel
-        queue_a, queue_b, queue_c = self.perform_txn_work(channel, "tx-autorollback-a", "tx-autorollback-b", "tx-autorollback-c")
+        channel2 = self.client.channel(2)
+        channel2.session_open()
+        queue_a, queue_b, queue_c = self.perform_txn_work(channel2, "tx-autorollback-a", "tx-autorollback-b", "tx-autorollback-c")
 
         for q in [queue_a, queue_b, queue_c]:
             try:
@@ -86,7 +87,18 @@ class TxTests(TestBase):
                 self.fail("Got unexpected message: " + extra.content.body)
             except Empty: None
 
-        channel.tx_rollback()
+        channel2.session_close()
+        channel = self.channel
+        channel.tx_select()
+
+        self.subscribe(channel, queue="tx-autorollback-a", destination="qa", confirm_mode=1)
+        queue_a = self.client.queue("qa")
+
+        self.subscribe(channel, queue="tx-autorollback-b", destination="qb", confirm_mode=1)
+        queue_b = self.client.queue("qb")
+
+        self.subscribe(channel, queue="tx-autorollback-c", destination="qc", confirm_mode=1)
+        queue_c = self.client.queue("qc")
 
         #check results
         for i in range(1, 5):
@@ -124,8 +136,17 @@ class TxTests(TestBase):
                 self.fail("Got unexpected message: " + extra.content.body)
             except Empty: None
 
+        #stop subscriptions (ensures no delivery occurs during rollback as messages are requeued)
+        for d in ["sub_a", "sub_b", "sub_c"]:
+            channel.message_stop(destination=d)
+
         channel.tx_rollback()
 
+        #restart susbcriptions
+        for d in ["sub_a", "sub_b", "sub_c"]:
+            channel.message_flow(destination=d, unit=0, value=0xFFFFFFFF)
+            channel.message_flow(destination=d, unit=1, value=0xFFFFFFFF)
+
         #check results
         for i in range(1, 5):
             msg = queue_a.get(timeout=1)