1 files changed, 85 insertions, 57 deletions

diff --git a/qpid/cpp/src/qpid/ha/QueueReplicator.cpp b/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
index 6aff4879e3..dbed7e1537 100644
--- a/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
+++ b/qpid/cpp/src/qpid/ha/QueueReplicator.cpp
@@ -19,6 +19,7 @@
  *
  */
 
+#include "Counter.h"
 #include "QueueReplicator.h"
 #include "ReplicatingSubscription.h"
 #include "qpid/broker/Bridge.h"
@@ -30,6 +31,7 @@
 #include "qpid/framing/SequenceSet.h"
 #include "qpid/framing/FieldTable.h"
 #include "qpid/log/Statement.h"
+#include "qpid/Msg.h"
 #include <boost/shared_ptr.hpp>
 
 namespace {
@@ -43,25 +45,38 @@ namespace ha {
 using namespace broker;
 using namespace framing;
 
-const std::string QueueReplicator::DEQUEUE_EVENT_KEY("qpid.dequeue-event");
-const std::string QueueReplicator::POSITION_EVENT_KEY("qpid.position-event");
+const std::string QPID_HA_EVENT_PREFIX("qpid.ha-event:");
+const std::string QueueReplicator::DEQUEUE_EVENT_KEY(QPID_HA_EVENT_PREFIX+"dequeue");
+const std::string QueueReplicator::POSITION_EVENT_KEY(QPID_HA_EVENT_PREFIX+"position");
 
 std::string QueueReplicator::replicatorName(const std::string& queueName) {
     return QPID_REPLICATOR_ + queueName;
 }
 
-QueueReplicator::QueueReplicator(boost::shared_ptr<Queue> q, boost::shared_ptr<Link> l)
-    : Exchange(replicatorName(q->getName()), 0, q->getBroker()), queue(q), link(l)
+bool QueueReplicator::isEventKey(const std::string key) {
+    const std::string& prefix = QPID_HA_EVENT_PREFIX;
+    bool ret = key.size() > prefix.size() && key.compare(0, prefix.size(), prefix) == 0;
+    return ret;
+}
+
+QueueReplicator::QueueReplicator(const BrokerInfo& info,
+                                 boost::shared_ptr<Queue> q,
+                                 boost::shared_ptr<Link> l)
+    : Exchange(replicatorName(q->getName()), 0, q->getBroker()),
+      logPrefix("Backup queue "+q->getName()+": "),
+      queue(q), link(l), brokerInfo(info)
 {
-    logPrefix = "HA: Backup " + queue->getName() + ": ";
-    QPID_LOG(info, logPrefix << "Created, settings: " << q->getSettings());
+    Uuid uuid(true);
+    bridgeName = replicatorName(q->getName()) + std::string(".") + uuid.str();
 }
 
 // This must be separate from the constructor so we can call shared_from_this.
 void QueueReplicator::activate() {
-    // Note this may create a new bridge or use an existing one.
+    sys::Mutex::ScopedLock l(lock);
+    std::pair<Bridge::shared_ptr, bool> result =
     queue->getBroker()->getLinks().declare(
-        link->getHost(), link->getPort(),
+        bridgeName,
+        *link,
         false,              // durable
         queue->getName(),   // src
         getName(),          // dest
@@ -74,47 +89,49 @@ void QueueReplicator::activate() {
         0,                  // sync?
         // Include shared_ptr to self to ensure we are not deleted
         // before initializeBridge is called.
-        boost::bind(&QueueReplicator::initializeBridge, this, _1, _2, shared_from_this())
+        boost::bind(&QueueReplicator::initializeBridge, shared_from_this(), _1, _2)
     );
+    bridge = result.first;
 }
 
 QueueReplicator::~QueueReplicator() {}
 
 void QueueReplicator::deactivate() {
+    // destroy the route
     sys::Mutex::ScopedLock l(lock);
-    queue->getBroker()->getLinks().destroy(
-        link->getHost(), link->getPort(), queue->getName(), getName(), string());
-    QPID_LOG(debug, logPrefix << "Deactivated bridge " << bridgeName);
+    if (bridge) {
+        bridge->close();
+        bridge.reset();
+        QPID_LOG(debug, logPrefix << "Deactivated bridge " << bridgeName);
+    }
 }
 
 // Called in a broker connection thread when the bridge is created.
-// shared_ptr to self ensures we are not deleted before initializeBridge is called.
-void QueueReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionHandler,
-                                       boost::shared_ptr<QueueReplicator> /*self*/) {
+void QueueReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionHandler) {
     sys::Mutex::ScopedLock l(lock);
-    bridgeName = bridge.getName();
-    framing::AMQP_ServerProxy peer(sessionHandler.out);
+    AMQP_ServerProxy peer(sessionHandler.out);
     const qmf::org::apache::qpid::broker::ArgsLinkBridge& args(bridge.getArgs());
-    framing::FieldTable settings;
-
-    // FIXME aconway 2011-12-09: Failover optimization removed.
-    // There was code here to re-use messages already on the backup
-    // during fail-over. This optimization was removed to simplify
-    // the logic till we get the basic replication stable, it
-    // can be re-introduced later. Last revision with the optimization:
-    // r1213258 | QPID-3603: Fix QueueReplicator subscription parameters.
-
-    // Clear out any old messages, reset the queue to start replicating fresh.
-    queue->purge();
-    queue->setPosition(0);
-
+    FieldTable settings;
     settings.setInt(ReplicatingSubscription::QPID_REPLICATING_SUBSCRIPTION, 1);
-    // TODO aconway 2011-12-19: optimize.
-    settings.setInt(QPID_SYNC_FREQUENCY, 1);
-    peer.getMessage().subscribe(args.i_src, args.i_dest, 0/*accept-explicit*/, 1/*not-acquired*/, false/*exclusive*/, "", 0, settings);
+    settings.setInt(QPID_SYNC_FREQUENCY, 1); // FIXME aconway 2012-05-22: optimize?
+    settings.setInt(ReplicatingSubscription::QPID_BACK,
+                    queue->getPosition());
+    settings.setTable(ReplicatingSubscription::QPID_BROKER_INFO,
+                      brokerInfo.asFieldTable());
+    SequenceNumber front;
+    if (ReplicatingSubscription::getFront(*queue, front))
+        settings.setInt(ReplicatingSubscription::QPID_FRONT, front);
+    peer.getMessage().subscribe(
+        args.i_src, args.i_dest, 0/*accept-explicit*/, 1/*not-acquired*/,
+        false/*exclusive*/, "", 0, settings);
+    // FIXME aconway 2012-05-22: use a finite credit window
     peer.getMessage().flow(getName(), 0, 0xFFFFFFFF);
     peer.getMessage().flow(getName(), 1, 0xFFFFFFFF);
-    QPID_LOG(debug, logPrefix << "Activated bridge " << bridgeName);
+
+    qpid::Address primary;
+    link->getRemoteAddress(primary);
+    QPID_LOG(info, logPrefix << "Connected to " << primary << "(" << bridgeName << ")");
+    QPID_LOG(trace, logPrefix << "Subscription settings: " << settings);
 }
 
 namespace {
@@ -128,35 +145,46 @@ template <class T> T decodeContent(Message& m) {
 }
 }
 
-void QueueReplicator::dequeue(SequenceNumber n,  const sys::Mutex::ScopedLock&) {
+void QueueReplicator::dequeue(SequenceNumber n, sys::Mutex::ScopedLock&) {
     // Thread safe: only calls thread safe Queue functions.
-    if (queue->getPosition() >= n) { // Ignore messages we  haven't reached yet
-        QueuedMessage message;
-        if (queue->acquireMessageAt(n, message))
-            queue->dequeue(0, message);
-    }
+    QueuedMessage message;
+    if (queue->acquireMessageAt(n, message))
+        queue->dequeue(0, message);
 }
 
 // Called in connection thread of the queues bridge to primary.
 void QueueReplicator::route(Deliverable& msg)
 {
-    const std::string& key = msg.getMessage().getRoutingKey();
-    sys::Mutex::ScopedLock l(lock);
-    if (key == DEQUEUE_EVENT_KEY) {
-        SequenceSet dequeues = decodeContent<SequenceSet>(msg.getMessage());
-        QPID_LOG(trace, logPrefix << "Dequeue: " << dequeues);
-        //TODO: should be able to optimise the following
-        for (SequenceSet::iterator i = dequeues.begin(); i != dequeues.end(); i++)
-            dequeue(*i, l);
-    } else if (key == POSITION_EVENT_KEY) {
-        SequenceNumber position = decodeContent<SequenceNumber>(msg.getMessage());
-        QPID_LOG(trace, logPrefix << "Position moved from " << queue->getPosition()
-                 << " to " << position);
-        assert(queue->getPosition() <= position);
-        queue->setPosition(position);
-    } else {
-        msg.deliverTo(queue);
-        QPID_LOG(trace, logPrefix << "Enqueued message " << queue->getPosition());
+    try {
+        const std::string& key = msg.getMessage().getRoutingKey();
+        sys::Mutex::ScopedLock l(lock);
+        if (!isEventKey(key)) {
+            msg.deliverTo(queue);
+            // We are on a backup so the queue is not modified except via this.
+            QPID_LOG(trace, logPrefix << "Enqueued message " << queue->getPosition());
+        }
+        else if (key == DEQUEUE_EVENT_KEY) {
+            SequenceSet dequeues = decodeContent<SequenceSet>(msg.getMessage());
+            QPID_LOG(trace, logPrefix << "Dequeue: " << dequeues);
+            //TODO: should be able to optimise the following
+            for (SequenceSet::iterator i = dequeues.begin(); i != dequeues.end(); i++)
+                dequeue(*i, l);
+        }
+        else if (key == POSITION_EVENT_KEY) {
+            SequenceNumber position = decodeContent<SequenceNumber>(msg.getMessage());
+            QPID_LOG(trace, logPrefix << "Position moved from " << queue->getPosition()
+                     << " to " << position);
+            // Verify that there are no messages after the new position in the queue.
+            SequenceNumber next;
+            if (ReplicatingSubscription::getNext(*queue, position, next))
+                throw Exception("Invalid position move, preceeds messages");
+            queue->setPosition(position);
+        }
+        // Ignore unknown event keys, may be introduced in later versions.
+    }
+    catch (const std::exception& e) {
+        QPID_LOG(critical, logPrefix << "Replication failed: " << e.what());
+        throw;
     }
 }