QPID-4327: HA clean up transaction artifacts at end of TX.

- Backups delete transactions on failover.
- TxReplicator cancel subscriptions when transaction is finished.
- TxReplicator rollback if destroyed prematurely.
- Handle special case of no backups for a tx.
- ha_tests.py: new and modified tests to cover the new functionality.

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

1 files changed, 13 insertions, 0 deletions

diff --git a/cpp/src/qpid/ha/QueueReplicator.cpp b/cpp/src/qpid/ha/QueueReplicator.cpp
index ff32dfef16..b4bbb3a0c4 100644
--- a/cpp/src/qpid/ha/QueueReplicator.cpp
+++ b/cpp/src/qpid/ha/QueueReplicator.cpp
@@ -169,6 +169,11 @@ void QueueReplicator::activate() {
         boost::shared_ptr<QueueObserver>(new QueueObserver(shared_from_this())));
 }
 
+void QueueReplicator::disconnect() {
+    Mutex::ScopedLock l(lock);
+    sessionHandler = 0;
+}
+
 QueueReplicator::~QueueReplicator() {}
 
 // Called from Queue::destroyed()
@@ -220,6 +225,14 @@ void QueueReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionHa
     QPID_LOG(trace, logPrefix << "Subscription arguments: " << arguments);
 }
 
+void QueueReplicator::cancel(Mutex::ScopedLock&) {
+    if (sessionHandler) {
+        // Cancel the replicating subscription.
+        AMQP_ServerProxy peer(sessionHandler->out);
+        peer.getMessage().cancel(getName());
+    }
+}
+
 namespace {
 template <class T> T decodeContent(Message& m) {
     std::string content = m.getContent();