Add support for array type to c++ (and python, decode only for now)

Change the type of the in-doubt field in dtx-coordination.recover to an array (to bring in line with amqp spec)



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

3 files changed, 61 insertions, 12 deletions

diff --git a/python/qpid/codec.py b/python/qpid/codec.py
index 18928698a5..6399ad2a5d 100644
--- a/python/qpid/codec.py
+++ b/python/qpid/codec.py
@@ -510,6 +510,23 @@ class Codec:
     type = self.spec.structs[codec.decode_short()]
     return codec.decode_struct_body(type)
 
+  def decode_array(self):
+    size = self.decode_long()
+    code = self.decode_octet()
+    count = self.decode_long()
+    result = []
+    for i in range(0, count):
+      if self.types.has_key(code):
+        value = self.decode(self.types[code])
+      else:
+        w = width(code)
+        if fixed(code):
+          value = self.read(w)
+        else:
+          value = self.read(self.dec_num(w))
+      result.append(value)
+    return result
+
 def fixed(code):
   return (code >> 6) != 2
 
diff --git a/python/qpid/spec.py b/python/qpid/spec.py
index 31c79276aa..4d0497bced 100644
--- a/python/qpid/spec.py
+++ b/python/qpid/spec.py
@@ -255,6 +255,7 @@ class Method(Metadata):
               "shortstr": "",
               "longstr": "",
               "table": {},
+              "array": [],
               "octet": 0,
               "short": 0,
               "long": 0,
diff --git a/python/tests_0-10/dtx.py b/python/tests_0-10/dtx.py
index 8fdd32c2f5..5ee4dd4c16 100644
--- a/python/tests_0-10/dtx.py
+++ b/python/tests_0-10/dtx.py
@@ -304,6 +304,38 @@ class DtxTests(TestBase):
         self.assertMessageId("a", "two")
         self.assertMessageId("b", "one")
 
+    def test_suspend_start_end_resume(self):        
+        """
+        Test suspension and resumption of an association with work
+        done on another transaction when the first transaction is
+        suspended
+        """
+        channel = self.channel
+        channel.dtx_demarcation_select()
+
+        #setup
+        channel.queue_declare(queue="one", exclusive=True, auto_delete=True)
+        channel.queue_declare(queue="two", exclusive=True, auto_delete=True)
+        channel.message_transfer(content=Content(properties={'routing_key':"one", 'message_id':"a"}, body="DtxMessage"))
+        channel.message_transfer(content=Content(properties={'routing_key':"two", 'message_id':"b"}, body="DtxMessage"))
+
+        tx = self.xid("dummy")
+
+        channel.dtx_demarcation_start(xid=tx)
+        self.swap(channel, "one", "two")#swap 'a' from 'one' to 'two'
+        channel.dtx_demarcation_end(xid=tx, suspend=True)
+
+        channel.dtx_demarcation_start(xid=tx, resume=True)
+        self.swap(channel, "two", "one")#swap 'b' from 'two' to 'one'
+        channel.dtx_demarcation_end(xid=tx)
+        
+        #commit and check
+        channel.dtx_coordination_commit(xid=tx, one_phase=True)
+        self.assertMessageCount(1, "one")
+        self.assertMessageCount(1, "two")
+        self.assertMessageId("a", "two")
+        self.assertMessageId("b", "one")
+
     def test_end_suspend_and_fail(self):
         """        
         Verify that the correct error is signalled if the suspend and
@@ -538,18 +570,7 @@ class DtxTests(TestBase):
             else:    
                 channel.dtx_coordination_rollback(xid=tx)
 
-        indoubt = channel.dtx_coordination_recover().in_doubt
-        #convert indoubt table to a list of xids (note: this will change for 0-10)
-        data = indoubt["xids"]
-        xids = []
-        pos = 0
-        while pos < len(data):
-            size = unpack("!B", data[pos])[0]
-            start = pos + 1
-            end = start + size
-            xid = data[start:end]
-            xids.append(xid)
-            pos = end
+        xids = channel.dtx_coordination_recover().in_doubt
         
         #rollback the prepared transactions returned by recover
         for x in xids:
@@ -567,6 +588,16 @@ class DtxTests(TestBase):
                 channel.dtx_coordination_rollback(xid=x)            
             self.fail("Recovered xids not as expected. missing: %s; extra: %s" % (missing, extra))
 
+    def test_bad_resume(self):
+        """
+        Test that a resume on a session not selected for use with dtx fails
+        """
+        channel = self.channel
+        try:
+            channel.dtx_demarcation_start(resume=True)
+        except Closed, e:
+            self.assertConnectionException(503, e.args[0])
+
     def xid(self, txid):
         DtxTests.tx_counter += 1
         branchqual = "v%s" % DtxTests.tx_counter