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

4 files changed, 13 insertions, 4 deletions

diff --git a/cpp/src/qpid/replication/ReplicatingEventListener.cpp b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
index 8aa47999cc..e3990a13cc 100644
--- a/cpp/src/qpid/replication/ReplicatingEventListener.cpp
+++ b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
@@ -57,7 +57,6 @@ void ReplicatingEventListener::deliverDequeueMessage(const QueuedMessage& dequeu
 {
     FieldTable headers;
     headers.setString(REPLICATION_TARGET_QUEUE, dequeued.queue->getName());
-    headers.setInt(REPLICATION_EVENT_SEQNO, ++sequence);
     headers.setInt(REPLICATION_EVENT_TYPE, DEQUEUE);
     headers.setInt(DEQUEUED_MESSAGE_POSITION, dequeued.position);
     boost::intrusive_ptr<Message> msg(createMessage(headers));
@@ -69,7 +68,6 @@ void ReplicatingEventListener::deliverEnqueueMessage(const QueuedMessage& enqueu
     boost::intrusive_ptr<Message> msg(cloneMessage(*(enqueued.queue), enqueued.payload));
     FieldTable& headers = msg->getProperties<MessageProperties>()->getApplicationHeaders();
     headers.setString(REPLICATION_TARGET_QUEUE, enqueued.queue->getName());
-    headers.setInt(REPLICATION_EVENT_SEQNO, ++sequence);
     headers.setInt(REPLICATION_EVENT_TYPE, ENQUEUE);
     queue->deliver(msg);
 }
@@ -138,6 +136,7 @@ void ReplicatingEventListener::initialize(Plugin::Target& target)
               queue = broker->getQueues().find(options.queue);
           }
           if (queue) {
+              queue->insertSequenceNumbers(REPLICATION_EVENT_SEQNO);
               QueueEvents::EventListener callback = boost::bind(&ReplicatingEventListener::handle, this, _1);
               broker->getQueueEvents().registerListener(options.name, callback);
               QPID_LOG(info, "Registered replicating queue event listener");
diff --git a/cpp/src/qpid/replication/ReplicatingEventListener.h b/cpp/src/qpid/replication/ReplicatingEventListener.h
index d302755704..3d8f23e7ac 100644
--- a/cpp/src/qpid/replication/ReplicatingEventListener.h
+++ b/cpp/src/qpid/replication/ReplicatingEventListener.h
@@ -58,7 +58,6 @@ class ReplicatingEventListener : public Plugin
 
     PluginOptions options;    
     qpid::broker::Queue::shared_ptr queue;
-    qpid::framing::SequenceNumber sequence;
 
     void deliverDequeueMessage(const qpid::broker::QueuedMessage& enqueued);
     void deliverEnqueueMessage(const qpid::broker::QueuedMessage& enqueued);
diff --git a/cpp/src/qpid/replication/ReplicationExchange.cpp b/cpp/src/qpid/replication/ReplicationExchange.cpp
index c35c6c2cd5..88c94ad7ba 100644
--- a/cpp/src/qpid/replication/ReplicationExchange.cpp
+++ b/cpp/src/qpid/replication/ReplicationExchange.cpp
@@ -34,11 +34,13 @@ using namespace qpid::broker;
 using namespace qpid::framing;
 using namespace qpid::replication::constants;
 
+const std::string SEQUENCE_VALUE("qpid.replication-event.sequence");
 ReplicationExchange::ReplicationExchange(const std::string& name, bool durable, 
                                          const FieldTable& args,
                                          QueueRegistry& qr,
                                          Manageable* parent) 
-    : Exchange(name, durable, args, parent), queues(qr), init(false) {}
+    : Exchange(name, durable, args, parent), queues(qr), sequence(args.getAsInt64(SEQUENCE_VALUE)), init(false)
+ {}
 
 std::string ReplicationExchange::getType() const { return typeName; }            
 
@@ -135,6 +137,13 @@ bool ReplicationExchange::isBound(Queue::shared_ptr /*queue*/, const string* con
 const std::string ReplicationExchange::typeName("replication");
 
 
+void ReplicationExchange::encode(Buffer& buffer) const
+{
+    args.setInt64(std::string(SEQUENCE_VALUE), sequence);
+    Exchange::encode(buffer);
+}
+
+
 struct ReplicationExchangePlugin : Plugin
 {
     Broker* broker;
diff --git a/cpp/src/qpid/replication/ReplicationExchange.h b/cpp/src/qpid/replication/ReplicationExchange.h
index 897e4a954e..4cc45ed5f5 100644
--- a/cpp/src/qpid/replication/ReplicationExchange.h
+++ b/cpp/src/qpid/replication/ReplicationExchange.h
@@ -22,6 +22,7 @@
  *
  */
 #include "qpid/broker/Exchange.h"
+#include "qpid/framing/Buffer.h"
 #include "qpid/framing/SequenceNumber.h"
 
 namespace qpid {
@@ -58,6 +59,7 @@ class ReplicationExchange : public qpid::broker::Exchange
     bool isDuplicate(const qpid::framing::FieldTable* args);
     void handleEnqueueEvent(const qpid::framing::FieldTable* args, qpid::broker::Deliverable& msg);
     void handleDequeueEvent(const qpid::framing::FieldTable* args);
+    void encode(framing::Buffer& buffer) const;
 };
 }} // namespace qpid::replication