From c9fc98ae80a5d8c4f58541f9738aa975723ff3d6 Mon Sep 17 00:00:00 2001 From: Alan Conway Date: Tue, 12 Jun 2012 21:19:48 +0000 Subject: QPID-3603: Separate QueueGuard from ReplicatingSubscription. QueueGuard: implements QueueObserver to delay completion of new messages. ReplicatingSubscription: Implements subscription, sends messages & events to backup. These were previously combined as one. QueueGuard is now separated out so that it can be created before the ReplicatingSubscription, in anticipation of an expected backup connecting. This is needed for 2 reasons: - new queues must be guarded until they are backuped up. - after a failover, all queues must be guarded until backups are ready. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1349538 13f79535-47bb-0310-9956-ffa450edef68 --- cpp/src/qpid/ha/QueueReplicator.cpp | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) (limited to 'cpp/src/qpid/ha/QueueReplicator.cpp') diff --git a/cpp/src/qpid/ha/QueueReplicator.cpp b/cpp/src/qpid/ha/QueueReplicator.cpp index 58c5e452d7..4d12015008 100644 --- a/cpp/src/qpid/ha/QueueReplicator.cpp +++ b/cpp/src/qpid/ha/QueueReplicator.cpp @@ -21,7 +21,6 @@ #include "Counter.h" #include "QueueReplicator.h" -#include "HaBroker.h" #include "ReplicatingSubscription.h" #include "qpid/broker/Bridge.h" #include "qpid/broker/Broker.h" @@ -60,15 +59,15 @@ bool QueueReplicator::isEventKey(const std::string key) { return ret; } -QueueReplicator::QueueReplicator(HaBroker& hb, +QueueReplicator::QueueReplicator(const BrokerInfo& info, boost::shared_ptr q, boost::shared_ptr l) : Exchange(replicatorName(q->getName()), 0, q->getBroker()), - haBroker(hb), logPrefix(hb), queue(q), link(l) + logPrefix("HA backup of "+q->getName()+": "), + queue(q), link(l), brokerInfo(info) { Uuid uuid(true); bridgeName = replicatorName(q->getName()) + std::string(".") + uuid.str(); - logPrefix.setMessage(q->getName()); QPID_LOG(info, logPrefix << "Created"); } @@ -119,7 +118,7 @@ void QueueReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionHa settings.setInt(ReplicatingSubscription::QPID_HIGH_SEQUENCE_NUMBER, queue->getPosition()); settings.setTable(ReplicatingSubscription::QPID_BROKER_INFO, - haBroker.getBrokerInfo().asFieldTable()); + brokerInfo.asFieldTable()); SequenceNumber front; if (ReplicatingSubscription::getFront(*queue, front)) settings.setInt(ReplicatingSubscription::QPID_LOW_SEQUENCE_NUMBER, front); @@ -143,7 +142,7 @@ template 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; -- cgit v1.2.1