QPID-1721: Fixes for replication between clusters when new members are added

* suppress event generation during node catch up 
* ensure sequence counters used for duplicate detection are synchronised in both primary and dr clusters when new members join
* connect queue with the event manager within queue registry rather than adapter as the latter path is not used for catchup



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

1 files changed, 16 insertions, 3 deletions

diff --git a/cpp/src/qpid/broker/QueueEvents.cpp b/cpp/src/qpid/broker/QueueEvents.cpp
index a6517e1bfe..7525e4cb76 100644
--- a/cpp/src/qpid/broker/QueueEvents.cpp
+++ b/cpp/src/qpid/broker/QueueEvents.cpp
@@ -20,12 +20,13 @@
  */
 #include "QueueEvents.h"
 #include "qpid/Exception.h"
+#include "qpid/log/Statement.h"
 
 namespace qpid {
 namespace broker {
 
 QueueEvents::QueueEvents(const boost::shared_ptr<sys::Poller>& poller) : 
-    eventQueue(boost::bind(&QueueEvents::handle, this, _1), poller) 
+    eventQueue(boost::bind(&QueueEvents::handle, this, _1), poller), enabled(true) 
 {
     eventQueue.start();
 }
@@ -37,12 +38,12 @@ QueueEvents::~QueueEvents()
 
 void QueueEvents::enqueued(const QueuedMessage& m)
 {
-    eventQueue.push(Event(ENQUEUE, m));
+    if (enabled) eventQueue.push(Event(ENQUEUE, m));
 }
 
 void QueueEvents::dequeued(const QueuedMessage& m)
 {
-    eventQueue.push(Event(DEQUEUE, m));
+    if (enabled) eventQueue.push(Event(DEQUEUE, m));
 }
 
 void QueueEvents::registerListener(const std::string& id, const EventListener& listener)
@@ -81,6 +82,18 @@ void QueueEvents::shutdown()
     if (!eventQueue.empty() && !listeners.empty()) eventQueue.shutdown();
 }
 
+void QueueEvents::enable()
+{
+    enabled = true;
+    QPID_LOG(debug, "Queue events enabled");
+}
+
+void QueueEvents::disable()
+{
+    enabled = false;
+    QPID_LOG(debug, "Queue events disabled");
+}
+
 QueueEvents::Event::Event(EventType t, const QueuedMessage& m) : type(t), msg(m) {}