QPID-4327: HA TX transactions: basic replication.

On primary a PrimaryTxObserver observes a transaction's TxBuffer and generates
transaction events on a tx-replication-queue. On the backup a TxReplicator
receives the events and constructs a TxBuffer equivalent to the one in the
primary.

Unfinished:
- Primary does not wait for backups to prepare() before committing.
- All connected backups are assumed to be in the transaction, there are race
  conditions around brokers joining/leavinv where this assumption is invalid.
- Need more tests.

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

1 files changed, 47 insertions, 23 deletions

diff --git a/qpid/cpp/src/tests/BrokerFixture.h b/qpid/cpp/src/tests/BrokerFixture.h
index bc23867ee1..b7b0f9d34b 100644
--- a/qpid/cpp/src/tests/BrokerFixture.h
+++ b/qpid/cpp/src/tests/BrokerFixture.h
@@ -42,11 +42,45 @@ namespace tests {
 struct  BrokerFixture : private boost::noncopyable {
     typedef qpid::broker::Broker Broker;
     typedef boost::intrusive_ptr<Broker> BrokerPtr;
+    typedef std::vector<std::string> Args;
 
     BrokerPtr broker;
+    uint16_t port;
     qpid::sys::Thread brokerThread;
 
-    BrokerFixture(Broker::Options opts=Broker::Options(), bool enableMgmt=false) {
+    BrokerFixture(const Args& args=Args(), const Broker::Options& opts=Broker::Options(),
+                  bool isExternalPort_=false, uint16_t externalPort_=0)
+    {
+        init(args, opts, isExternalPort_, externalPort_);
+    }
+
+    BrokerFixture(const Broker::Options& opts,
+                  bool isExternalPort_=false, uint16_t externalPort_=0)
+    {
+        init(Args(), opts, isExternalPort_, externalPort_);
+    }
+
+    void shutdownBroker() {
+        if (broker) {
+            broker->shutdown();
+            brokerThread.join();
+            broker = BrokerPtr();
+        }
+    }
+
+    ~BrokerFixture() {  shutdownBroker(); }
+
+    /** Open a connection to the broker. */
+    void open(qpid::client::Connection& c) {
+        c.open("localhost", getPort());
+    }
+
+    uint16_t getPort() { return port; }
+
+  private:
+    void init(const Args& args, Broker::Options opts,
+              bool isExternalPort=false, uint16_t externalPort=0)
+    {
         // Keep the tests quiet unless logging env. vars have been set by user.
         if (!::getenv("QPID_LOG_ENABLE") && !::getenv("QPID_TRACE")) {
             qpid::log::Options logOpts;
@@ -55,38 +89,28 @@ struct  BrokerFixture : private boost::noncopyable {
             logOpts.selectors.push_back("error+");
             qpid::log::Logger::instance().configure(logOpts);
         }
+        // Default options, may be over-ridden when we parse args.
         opts.port=0;
         opts.listenInterfaces.push_back("127.0.0.1");
-        // Management doesn't play well with multiple in-process brokers.
-        opts.enableMgmt=enableMgmt;
         opts.workerThreads=1;
         opts.dataDir="";
         opts.auth=false;
+
+        // Argument parsing
+        std::vector<const char*> argv(args.size());
+        std::transform(args.begin(), args.end(), argv.begin(),
+                       boost::bind(&std::string::c_str, _1));
+        Plugin::addOptions(opts);
+        opts.parse(argv.size(), &argv[0]);
         broker = Broker::create(opts);
         // TODO aconway 2007-12-05: At one point BrokerFixture
         // tests could hang in Connection ctor if the following
         // line is removed. This may not be an issue anymore.
         broker->accept();
-        broker->getPort(qpid::broker::Broker::TCP_TRANSPORT);
+        if (isExternalPort) port = externalPort;
+        else port = broker->getPort(qpid::broker::Broker::TCP_TRANSPORT);
         brokerThread = qpid::sys::Thread(*broker);
     };
-
-    void shutdownBroker() {
-        if (broker) {
-            broker->shutdown();
-            brokerThread.join();
-            broker = BrokerPtr();
-        }
-    }
-
-    ~BrokerFixture() {  shutdownBroker(); }
-
-    /** Open a connection to the broker. */
-    void open(qpid::client::Connection& c) {
-        c.open("localhost", broker->getPort(qpid::broker::Broker::TCP_TRANSPORT));
-    }
-
-    uint16_t getPort() { return broker->getPort(qpid::broker::Broker::TCP_TRANSPORT); }
 };
 
 /** Connection that opens in its constructor */
@@ -125,8 +149,8 @@ template <class ConnectionType, class SessionType=qpid::client::Session>
 struct  SessionFixtureT : BrokerFixture, ClientT<ConnectionType,SessionType> {
 
     SessionFixtureT(Broker::Options opts=Broker::Options()) :
-        BrokerFixture(opts),
-        ClientT<ConnectionType,SessionType>(broker->getPort(qpid::broker::Broker::TCP_TRANSPORT))
+        BrokerFixture(BrokerFixture::Args(), opts),
+        ClientT<ConnectionType,SessionType>(getPort())
     {}
 
 };