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, 5 insertions, 14 deletions

diff --git a/qpid/cpp/src/tests/test_store.cpp b/qpid/cpp/src/tests/test_store.cpp
index ee04dddd6a..14aee7b648 100644
--- a/qpid/cpp/src/tests/test_store.cpp
+++ b/qpid/cpp/src/tests/test_store.cpp
@@ -223,27 +223,18 @@ class TestStore : public NullMessageStore {
                  const boost::intrusive_ptr<PersistableMessage>& pmsg,
                  const PersistableQueue& queue)
     {
-        qpid::broker::amqp_0_10::MessageTransfer* msg =
-            dynamic_cast<qpid::broker::amqp_0_10::MessageTransfer*>(pmsg.get());
-        assert(msg);
-
         ostringstream o;
-        o << "<enqueue " << queue.getName() << " " << getContent(msg);
+        string data = getContent(pmsg);
+        o << "<enqueue " << queue.getName() << " " << data;
         if (tx) o << " tx=" << getId(*tx);
         o << ">";
         log(o.str());
 
         // Dump the message if there is a dump file.
         if (dump.get()) {
-            msg->getFrames().getMethod()->print(*dump);
-            *dump  << endl << "  ";
-            msg->getFrames().getHeaders()->print(*dump);
-            *dump << endl << "  ";
-            *dump << msg->getFrames().getContentSize() << endl;
+            *dump << "Message(" << data.size() << "): " << data << endl;
         }
         string logPrefix = "TestStore "+name+": ";
-        // Check the message for special instructions for this store.
-        string data = msg->getFrames().getContent();
         Action action(data);
         bool doComplete = true;
         if (action.index && action.executeIn(name)) {
@@ -258,7 +249,7 @@ class TestStore : public NullMessageStore {
                       QPID_LOG(error, logPrefix << "async-id needs argument: " << data);
                       break;
                   }
-                  asyncIds[action.args[0]] = msg;
+                  asyncIds[action.args[0]] = pmsg;
                   QPID_LOG(debug, logPrefix << "delayed completion " << action.args[0]);
                   doComplete = false;
                   break;
@@ -284,7 +275,7 @@ class TestStore : public NullMessageStore {
                 QPID_LOG(error, logPrefix << "unknown action: " << data);
             }
         }
-        if (doComplete) msg->enqueueComplete();
+        if (doComplete) pmsg->enqueueComplete();
     }
 
     void dequeue(TransactionContext* tx,