QPID-4327: Added TransactionObserver interface.

Added TransactionObserver interface, called at each point in a transaction's
lifecycle. Currently only a single observer can be associated with a
transaction.

Added startTx, startDtx to BrokerObserver so plugins can observe transactions
starting and set a TransactionObserver.

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

1 files changed, 45 insertions, 0 deletions

diff --git a/qpid/cpp/src/tests/ha_tests.py b/qpid/cpp/src/tests/ha_tests.py
index 293712fe80..de5dfb4b10 100755
--- a/qpid/cpp/src/tests/ha_tests.py
+++ b/qpid/cpp/src/tests/ha_tests.py
@@ -1287,6 +1287,51 @@ class StoreTests(BrokerTest):
         cluster[0].assert_browse("q2", ["hello", "end"])
         cluster[1].assert_browse_backup("q2", ["hello", "end"])
 
+class TransactionTests(BrokerTest):
+
+    def tx_simple_setup(self, broker):
+        """Start a transaction: receive 'foo' from 'a' and send 'bar' to 'b'"""
+        c = broker.connect()
+        c.session().sender("a;{create:always}").send("foo")
+        tx = c.session(transactional=True)
+        self.assertEqual("foo", tx.receiver("a").fetch(1).content)
+        tx.acknowledge();
+        tx.sender("b;{create:always}").send("bar")
+        return tx
+
+    def test_tx_simple_commit(self):
+        cluster = HaCluster(self, 2, args=["--log-enable=trace+:ha::"])
+        tx = self.tx_simple_setup(cluster[0])
+        tx.commit()
+        for b in cluster:
+            b.assert_browse_backup("a", [], msg=b)
+            b.assert_browse_backup("b", ["bar"], msg=b)
+
+    def test_tx_simple_rollback(self):
+        cluster = HaCluster(self, 2)
+        tx = self.tx_simple_setup(cluster[0])
+        tx.rollback()
+        for b in cluster:
+            b.assert_browse_backup("a", ["foo"], msg=b)
+            b.assert_browse_backup("b", [], msg=b)
+
+    def test_tx_simple_failover(self):
+        cluster = HaCluster(self, 2)
+        tx = self.tx_simple_setup(cluster[0])
+        cluster.bounce(0)       # Should cause roll-back
+        for b in cluster:
+            b.assert_browse_backup("a", ["foo"], msg=b)
+            b.assert_browse_backup("b", [], msg=b)
+
+    def test_tx_simple_join(self):
+        cluster = HaCluster(self, 2)
+        tx = self.tx_simple_setup(cluster[0])
+        cluster.bounce(1)       # Should catch up with tx
+        tx.commit()
+        for b in cluster:
+            b.assert_browse_backup("a", [], msg=b)
+            b.assert_browse_backup("b", ["bar"], msg=b)
+
 if __name__ == "__main__":
     outdir = "ha_tests.tmp"
     shutil.rmtree(outdir, True)