QPID-4327: HA TX transactions, blocking wait for prepare

Backups send prepare messages to primary, primary delays completion of prepare
till all are prepared (or there is a failure).

This is NOT the production solution - blocking could cause a deadlock. We need
to introduce asynchronous completion of prepare without blocking. This
interim solution allows testing on other aspects of TX support.

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

1 files changed, 10 insertions, 5 deletions

diff --git a/cpp/src/qpid/ha/QueueReplicator.cpp b/cpp/src/qpid/ha/QueueReplicator.cpp
index 28e9dc4120..9149567cf2 100644
--- a/cpp/src/qpid/ha/QueueReplicator.cpp
+++ b/cpp/src/qpid/ha/QueueReplicator.cpp
@@ -107,9 +107,13 @@ QueueReplicator::QueueReplicator(HaBroker& hb,
     : Exchange(replicatorName(q->getName()), 0, q->getBroker()),
       haBroker(hb),
       brokerInfo(hb.getBrokerInfo()),
+      link(l),
+      queue(q),
+      sessionHandler(0),
       logPrefix("Backup of "+q->getName()+": "),
-      queue(q), link(l), subscribed(false),
-      settings(hb.getSettings()), destroyed(false),
+      subscribed(false),
+      settings(hb.getSettings()),
+      destroyed(false),
       nextId(0), maxId(0)
 {
     args.setString(QPID_REPLICATE, printable(NONE).str());
@@ -176,22 +180,23 @@ void QueueReplicator::destroy() {
         if (destroyed) return;
         destroyed = true;
         QPID_LOG(debug, logPrefix << "Destroyed");
+        bridge2 = bridge;       // call close outside the lock.
         // Need to drop shared pointers to avoid pointer cycles keeping this in memory.
         queue.reset();
         link.reset();
         bridge.reset();
         getBroker()->getExchanges().destroy(getName());
-        bridge2 = bridge;
     }
     if (bridge2) bridge2->close(); // Outside of lock, avoid deadlock.
 }
 
 // Called in a broker connection thread when the bridge is created.
 // Note: called with the Link lock held.
-void QueueReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionHandler) {
+void QueueReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionHandler_) {
     Mutex::ScopedLock l(lock);
     if (destroyed) return;         // Already destroyed
-    AMQP_ServerProxy peer(sessionHandler.out);
+    sessionHandler = &sessionHandler_;
+    AMQP_ServerProxy peer(sessionHandler->out);
     const qmf::org::apache::qpid::broker::ArgsLinkBridge& args(bridge.getArgs());
     FieldTable arguments;
     arguments.setInt(ReplicatingSubscription::QPID_REPLICATING_SUBSCRIPTION, 1);