summaryrefslogtreecommitdiff
path: root/cpp/src
diff options
context:
space:
mode:
authorAlan Conway <aconway@apache.org>2012-07-05 19:57:25 +0000
committerAlan Conway <aconway@apache.org>2012-07-05 19:57:25 +0000
commit10832bb16a567b68189299a6e52dbf4ee6464744 (patch)
tree961618be9a7fb176bebbd1f11e8e083349a0bffb /cpp/src
parentec51c8a308cced792f6ff8d7f16876548f1cb73b (diff)
downloadqpid-python-10832bb16a567b68189299a6e52dbf4ee6464744.tar.gz
QPID-4085: HA message-loss race condition, handling replication event after response.
If the backup broker receives a declare event for a queue after receiving a queue response for the same queue, it removes the queue and replaces it with the new one from the reponse. Previously it did not remove the corresponding bridge so things fail when we attempt to create it. Corrected to remove the bridge also. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1357846 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'cpp/src')
-rw-r--r--cpp/src/qpid/ha/BrokerReplicator.cpp19
-rw-r--r--cpp/src/qpid/ha/BrokerReplicator.h1
-rw-r--r--cpp/src/qpid/ha/Primary.cpp4
-rw-r--r--cpp/src/qpid/ha/QueueReplicator.cpp2
4 files changed, 18 insertions, 8 deletions
diff --git a/cpp/src/qpid/ha/BrokerReplicator.cpp b/cpp/src/qpid/ha/BrokerReplicator.cpp
index 83c2eaa144..1fabff6a09 100644
--- a/cpp/src/qpid/ha/BrokerReplicator.cpp
+++ b/cpp/src/qpid/ha/BrokerReplicator.cpp
@@ -312,6 +312,7 @@ void BrokerReplicator::doEventQueueDeclare(Variant::Map& values) {
if (broker.getQueues().find(name)) {
QPID_LOG(warning, logPrefix << "Replacing exsiting queue: " << name);
broker.getQueues().destroy(name);
+ stopQueueReplicator(name);
}
std::pair<boost::shared_ptr<Queue>, bool> result =
broker.createQueue(
@@ -343,13 +344,7 @@ void BrokerReplicator::doEventQueueDelete(Variant::Map& values) {
boost::shared_ptr<Queue> queue = broker.getQueues().find(name);
if (queue && replicationTest.replicateLevel(queue->getSettings())) {
QPID_LOG(debug, logPrefix << "Queue delete event: " << name);
- boost::shared_ptr<QueueReplicator> qr = findQueueReplicator(name);
- if (qr) {
- qr->deactivate();
- // QueueReplicator's bridge is now queued for destruction but may not
- // actually be destroyed.
- broker.getExchanges().destroy(qr->getName());
- }
+ stopQueueReplicator(name);
broker.deleteQueue(name, values[USER].asString(), values[RHOST].asString());
}
}
@@ -563,6 +558,16 @@ void BrokerReplicator::startQueueReplicator(const boost::shared_ptr<Queue>& queu
}
}
+void BrokerReplicator::stopQueueReplicator(const std::string& name) {
+ boost::shared_ptr<QueueReplicator> qr = findQueueReplicator(name);
+ if (qr) {
+ qr->deactivate();
+ // QueueReplicator's bridge is now queued for destruction but may not
+ // actually be destroyed.
+ broker.getExchanges().destroy(qr->getName());
+ }
+}
+
bool BrokerReplicator::bind(boost::shared_ptr<Queue>, const string&, const framing::FieldTable*) { return false; }
bool BrokerReplicator::unbind(boost::shared_ptr<Queue>, const string&, const framing::FieldTable*) { return false; }
bool BrokerReplicator::isBound(boost::shared_ptr<Queue>, const string* const, const framing::FieldTable* const) { return false; }
diff --git a/cpp/src/qpid/ha/BrokerReplicator.h b/cpp/src/qpid/ha/BrokerReplicator.h
index f7439fe892..8289ad7e9d 100644
--- a/cpp/src/qpid/ha/BrokerReplicator.h
+++ b/cpp/src/qpid/ha/BrokerReplicator.h
@@ -94,6 +94,7 @@ class BrokerReplicator : public broker::Exchange,
QueueReplicatorPtr findQueueReplicator(const std::string& qname);
void startQueueReplicator(const boost::shared_ptr<broker::Queue>&);
+ void stopQueueReplicator(const std::string& name);
std::string logPrefix;
ReplicationTest replicationTest;
diff --git a/cpp/src/qpid/ha/Primary.cpp b/cpp/src/qpid/ha/Primary.cpp
index 5eb6b292f7..56598c2b5a 100644
--- a/cpp/src/qpid/ha/Primary.cpp
+++ b/cpp/src/qpid/ha/Primary.cpp
@@ -28,6 +28,7 @@
#include "qpid/assert.h"
#include "qpid/broker/Broker.h"
#include "qpid/broker/ConfigurationObserver.h"
+#include "qpid/broker/Connection.h"
#include "qpid/broker/Queue.h"
#include "qpid/framing/FieldTable.h"
#include "qpid/log/Statement.h"
@@ -161,6 +162,9 @@ void Primary::opened(broker::Connection& connection) {
}
haBroker.addBroker(info);
}
+ else
+ QPID_LOG(debug, logPrefix << "Accepted client connection "
+ << connection.getMgmtId())
}
void Primary::closed(broker::Connection& connection) {
diff --git a/cpp/src/qpid/ha/QueueReplicator.cpp b/cpp/src/qpid/ha/QueueReplicator.cpp
index dbed7e1537..70dd91f57f 100644
--- a/cpp/src/qpid/ha/QueueReplicator.cpp
+++ b/cpp/src/qpid/ha/QueueReplicator.cpp
@@ -94,7 +94,7 @@ void QueueReplicator::activate() {
bridge = result.first;
}
-QueueReplicator::~QueueReplicator() {}
+QueueReplicator::~QueueReplicator() { deactivate(); }
void QueueReplicator::deactivate() {
// destroy the route