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

6 files changed, 194 insertions, 3 deletions

diff --git a/qpid/cpp/src/tests/CMakeLists.txt b/qpid/cpp/src/tests/CMakeLists.txt
index 646374d692..a1d29c23a3 100644
--- a/qpid/cpp/src/tests/CMakeLists.txt
+++ b/qpid/cpp/src/tests/CMakeLists.txt
@@ -146,6 +146,7 @@ set(all_unit_tests
     TimerTest
     TopicExchangeTest
     TxBufferTest
+    TransactionObserverTest
     Url
     Uuid
     Variant
diff --git a/qpid/cpp/src/tests/TransactionObserverTest.cpp b/qpid/cpp/src/tests/TransactionObserverTest.cpp
new file mode 100644
index 0000000000..c284417e25
--- /dev/null
+++ b/qpid/cpp/src/tests/TransactionObserverTest.cpp
@@ -0,0 +1,143 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "unit_test.h"
+#include "test_tools.h"
+#include "MessagingFixture.h"
+#include "qpid/broker/BrokerObserver.h"
+#include "qpid/broker/TransactionObserver.h"
+#include "qpid/broker/TxBuffer.h"
+#include "qpid/broker/Queue.h"
+#include "qpid/ha/types.h"
+
+#include <boost/bind.hpp>
+#include <boost/function.hpp>
+#include <boost/make_shared.hpp>
+#include <boost/lexical_cast.hpp>
+#include <iostream>
+#include <vector>
+
+namespace qpid {
+namespace tests {
+
+using framing::SequenceSet;
+using  messaging::Message;
+
+using namespace boost::assign;
+using namespace boost;
+using namespace broker;
+using namespace std;
+using namespace messaging;
+using namespace types;
+
+QPID_AUTO_TEST_SUITE(TransactionalObserverTest)
+
+Message msg(string content) { return Message(content); }
+
+struct MockTransactionObserver : public TransactionObserver {
+    bool prep;
+    vector<string> events;
+
+    MockTransactionObserver(bool prep_=true) : prep(prep_) {}
+
+    void record(const string& e) { events.push_back(e); }
+
+    void enqueue(const shared_ptr<Queue>& q,  const broker::Message& m) {
+        record("enqueue "+q->getName()+" "+m.getContent());
+    }
+    void dequeue(const Queue::shared_ptr& q, SequenceNumber p, SequenceNumber r) {
+        record("dequeue "+q->getName()+" "+
+               lexical_cast<string>(p)+" "+lexical_cast<string>(r));
+    }
+    bool prepare() { record("prepare"); return prep; }
+    void commit() { record("commit"); }
+    void rollback() {record("rollback"); }
+};
+
+struct MockBrokerObserver : public BrokerObserver {
+    bool prep;
+    shared_ptr<MockTransactionObserver> tx;
+
+    MockBrokerObserver(bool prep_=true) : prep(prep_) {}
+
+    void startTx(const shared_ptr<TxBuffer>& buffer) {
+        tx = make_shared<MockTransactionObserver>(prep);
+        buffer->setObserver(tx);
+    }
+};
+
+Session simpleTxTransaction(MessagingFixture& fix) {
+    fix.session.createSender("q1;{create:always}").send(msg("foo")); // Not in TX
+    // Transaction with 1 enqueue and 1 dequeue.
+    Session txSession = fix.connection.createTransactionalSession();
+    BOOST_CHECK_EQUAL("foo", txSession.createReceiver("q1").fetch().getContent());
+    txSession.createSender("q2;{create:always}").send(msg("bar"));
+    return txSession;
+}
+
+QPID_AUTO_TEST_CASE(tesTxtCommit) {
+    MessagingFixture fix;
+    shared_ptr<MockBrokerObserver> brokerObserver(new MockBrokerObserver);
+    fix.broker->getBrokerObservers().add(brokerObserver);
+    Session txSession = simpleTxTransaction(fix);
+    txSession.commit();
+    // Note on ordering: observers see enqueues as they happen, but dequeues just
+    // before prepare.
+    BOOST_CHECK_EQUAL(
+        list_of<string>("enqueue q2 bar")("dequeue q1 1 0")("prepare")("commit"),
+        brokerObserver->tx->events
+    );
+}
+
+QPID_AUTO_TEST_CASE(testTxFail) {
+    MessagingFixture fix;
+    shared_ptr<MockBrokerObserver> brokerObserver(new MockBrokerObserver(false));
+    fix.broker->getBrokerObservers().add(brokerObserver);
+    Session txSession = simpleTxTransaction(fix);
+    try {
+        txSession.commit();
+        BOOST_FAIL("Expected exception");
+    } catch(...) {}
+
+    BOOST_CHECK_EQUAL(
+        list_of<string>("enqueue q2 bar")("dequeue q1 1 0")("prepare")("rollback"),
+        brokerObserver->tx->events
+    );
+}
+
+QPID_AUTO_TEST_CASE(testTxRollback) {
+    MessagingFixture fix;
+    shared_ptr<MockBrokerObserver> brokerObserver(new MockBrokerObserver(false));
+    fix.broker->getBrokerObservers().add(brokerObserver);
+    Session txSession = simpleTxTransaction(fix);
+    txSession.rollback();
+    // Note: The dequeue does not appear here. This is because TxAccepts
+    // (i.e. dequeues) are not enlisted until SemanticState::commit and are
+    // never enlisted if the transaction is rolled back.
+    BOOST_CHECK_EQUAL(
+        list_of<string>("enqueue q2 bar")("rollback"),
+        brokerObserver->tx->events
+    );
+}
+
+QPID_AUTO_TEST_SUITE_END()
+
+}} // namespace qpid::tests
diff --git a/qpid/cpp/src/tests/TxMocks.h b/qpid/cpp/src/tests/TxMocks.h
index bf21104f70..8b54e7484b 100644
--- a/qpid/cpp/src/tests/TxMocks.h
+++ b/qpid/cpp/src/tests/TxMocks.h
@@ -103,6 +103,9 @@ public:
         if(!debugName.empty()) std::cout << std::endl << "MockTxOp[" << debugName << "]::rollback()" << std::endl;
         actual.push_back(ROLLBACK);
     }
+
+    void callObserver(const boost::shared_ptr<TransactionObserver>&) {}
+
     MockTxOp& expectPrepare(){
         expected.push_back(PREPARE);
         return *this;
diff --git a/qpid/cpp/src/tests/brokertest.py b/qpid/cpp/src/tests/brokertest.py
index 03defddb58..286beb0258 100644
--- a/qpid/cpp/src/tests/brokertest.py
+++ b/qpid/cpp/src/tests/brokertest.py
@@ -78,7 +78,7 @@ def error_line(filename, n=1):
     except: return ""
     return ":\n" + "".join(result)
 
-def retry(function, timeout=10, delay=.01, max_delay=1):
+def retry(function, timeout=10, delay=.001, max_delay=1):
     """Call function until it returns a true value or timeout expires.
     Double the delay for each retry up to max_delay.
     Returns what function returns if true, None if timeout expires."""
@@ -398,7 +398,7 @@ def assert_browse(session, queue, expect_contents, timeout=0, transform=lambda m
     if msg: msg = "%s: %r != %r"%(msg, expect_contents, actual_contents)
     assert expect_contents == actual_contents, msg
 
-def assert_browse_retry(session, queue, expect_contents, timeout=1, delay=.01, transform=lambda m:m.content, msg="browse failed"):
+def assert_browse_retry(session, queue, expect_contents, timeout=1, delay=.001, transform=lambda m:m.content, msg="browse failed"):
     """Wait up to timeout for contents of queue to match expect_contents"""
     test = lambda: browse(session, queue, 0, transform=transform) == expect_contents
     retry(test, timeout, delay)
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)
diff --git a/qpid/cpp/src/tests/test_tools.h b/qpid/cpp/src/tests/test_tools.h
index de672f938a..7950a36913 100644
--- a/qpid/cpp/src/tests/test_tools.h
+++ b/qpid/cpp/src/tests/test_tools.h
@@ -23,7 +23,6 @@
 #include <limits.h>             // Include before boost/test headers.
 #include <boost/test/test_tools.hpp>
 #include <boost/assign/list_of.hpp>
-#include <boost/assign/list_of.hpp>
 #include <vector>
 #include <set>
 #include <ostream>