QPID-4710: [AMQP 1.0] Support for transactions in qpid::messaging C++ client.

Implements the "transactional retire and settle immediately" option for
transactions as specified in AMQP 1.0 in the qpid::messaging C++ client.

NOTE: Transactions over AMQP 1.0 require proton 0.9 or greater.  With older
versions, attempting a transactions over AMQP 1.0 will raise a link-detached
exception "Node not found: tx-transaction"

1. Added descriptor list to Variant with support in Encoder and PnData.

Required to support transactions, need to be able to create described lists.
Variant changes are source and binary compatible.

A Variant now has a Variant::List of descripors which can be numeric or string.
Nested descriptors are implemented by putting multiple descriptors in the list.

Other minor changes:
- Variant refactor: don't delete impl on every assignment.
- Add Variant constructors that take a string encoding.
  (new constructors, not defaulted arguments, so the change is binary and source compatible.)
- Growable buffer support for Encoder.
- Printing described Variant prints descriptors in form @descriptor value

2. Added transaction support to AMQP 1.0 client code

Added messaging/amqp/Transaction.h,cpp: transaction logic
- communicate with coordinator, send declare/dischange messages.
- add tx state info to transfers and acknowledgements.
- Sync session after discharge.
- A transactional session automatically acks any message retrieved by fetch/get
  to bring them into the transaction. This is consistent the 0-10 client.

Minor fixes to existing client code:
- Fix use of pn_drain API in C++ client to work with C++ and Java brokers.
- Make amqp::Exception derive from qpid::Exception

3. Fixes to existing broker code:

- Incoming.cpp fix: start async completion before processing message.
- Delay accept of dischage message till commit is complete.
- newSession - handle failover during session creation.

4. Added tests

interop_tests.py: transaction tests that can run against an external broker, see comments.

ha_tests.py: Enable transaction tests over AMQP 1.0.

Minor test fixes:
- brokertest.py don't set default logging if QPID_LOG env vars set.
- brokertest.py Pass kwargs to broker() create function.
- qpid-receive: capacity should never be larger than message count.
- Accept user:pass as well as user/pass in Url.
- brokertest.py: Always do a ready() check on all brokers.

If proton < 0.9 is used, transaction tests will be skipped or will downgrade to
the amqp0-10 protocol with a printed warning.

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

1 files changed, 12 insertions, 9 deletions

diff --git a/qpid/cpp/src/tests/ha_tests.py b/qpid/cpp/src/tests/ha_tests.py
index 1d475ebfe7..180831569f 100755
--- a/qpid/cpp/src/tests/ha_tests.py
+++ b/qpid/cpp/src/tests/ha_tests.py
@@ -1025,8 +1025,8 @@ class LongTests(HaBrokerTest):
              "--broker", brokers[0].host_port(),
              "--address", "q;{create:always}",
              "--messages=1000",
-             "--tx=10"
-             # TODO aconway 2014-02-21: can't use amqp1.0 for transactions yet
+             "--tx=10",
+             "--connection-options={protocol:%s}" % self.tx_protocol
              ])
         receiver = self.popen(
             ["qpid-receive",
@@ -1034,8 +1034,8 @@ class LongTests(HaBrokerTest):
              "--address", "q;{create:always}",
              "--messages=990",
              "--timeout=10",
-             "--tx=10"
-             # TODO aconway 2014-02-21: can't use amqp1.0 for transactions yet
+             "--tx=10",
+             "--connection-options={protocol:%s}" % self.tx_protocol
              ])
         self.assertEqual(sender.wait(), 0)
         self.assertEqual(receiver.wait(), 0)
@@ -1268,7 +1268,7 @@ class StoreTests(HaBrokerTest):
         """Verify that a backup erases queue data from store recovery before
         doing catch-up from the primary."""
         if self.check_skip(): return
-        cluster = HaCluster(self, 2, args=['--log-enable=trace+:ha', '--log-enable=trace+:Store'])
+        cluster = HaCluster(self, 2)
         sn = cluster[0].connect(heartbeat=HaBroker.heartbeat).session()
         s1 = sn.sender("q1;{create:always,node:{durable:true}}")
         for m in ["foo","bar"]: s1.send(qm.Message(m, durable=True))
@@ -1532,7 +1532,7 @@ class TransactionTests(HaBrokerTest):
             except qm.TransactionUnknown: pass
             for b in cluster: self.assert_tx_clean(b)
             try: tx.connection.close()
-            except TransactionUnknown: pass # Occasionally get exception on close.
+            except qm.TransactionUnknown: pass # Occasionally get exception on close.
         finally: l.restore()
 
     def test_tx_no_backups(self):
@@ -1622,17 +1622,20 @@ class TransactionTests(HaBrokerTest):
             import qpid_tests.broker_0_10
         except ImportError:
             raise Skipped("Tests not found")
-
         cluster = HaCluster(self, 3)
-        self.popen(["qpid-txtest", "-p%s"%cluster[0].port()]).assert_exit_ok()
+        if "QPID_PORT" in os.environ: del os.environ["QPID_PORT"]
+        self.popen(["qpid-txtest2", "--broker", cluster[0].host_port()]).assert_exit_ok()
+        print
         self.popen(["qpid-python-test",
                     "-m", "qpid_tests.broker_0_10",
+                    "-m", "qpid_tests.broker_1_0",
                     "-b", "localhost:%s"%(cluster[0].port()),
-                    "*.tx.*"]).assert_exit_ok()
+                    "*.tx.*"], stdout=None, stderr=None).assert_exit_ok()
 
 if __name__ == "__main__":
     qpid_ha_exec = os.getenv("QPID_HA_EXEC")
     if qpid_ha_exec and os.path.isfile(qpid_ha_exec):
+        BrokerTest.amqp_tx_warning()
         outdir = "ha_tests.tmp"
         shutil.rmtree(outdir, True)
         os.execvp("qpid-python-test",