Make replication of queue events not dependant on abs queue position, but rather use source queue sequence numbers. Also add in more error checking and tests

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

3 files changed, 23 insertions, 9 deletions

diff --git a/cpp/src/qpid/replication/ReplicatingEventListener.cpp b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
index 1a3ce1c069..b7d52372f4 100644
--- a/cpp/src/qpid/replication/ReplicatingEventListener.cpp
+++ b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
@@ -72,6 +72,7 @@ void ReplicatingEventListener::deliverEnqueueMessage(const QueuedMessage& enqueu
     FieldTable& headers = msg->getProperties<MessageProperties>()->getApplicationHeaders();
     headers.setString(REPLICATION_TARGET_QUEUE, enqueued.queue->getName());
     headers.setInt(REPLICATION_EVENT_TYPE, ENQUEUE);
+    headers.setInt(QUEUE_MESSAGE_POSITION,enqueued.position);
     route(msg);
 }
 
diff --git a/cpp/src/qpid/replication/ReplicationExchange.cpp b/cpp/src/qpid/replication/ReplicationExchange.cpp
index c0cc36efe3..053335e6ad 100644
--- a/cpp/src/qpid/replication/ReplicationExchange.cpp
+++ b/cpp/src/qpid/replication/ReplicationExchange.cpp
@@ -83,15 +83,27 @@ void ReplicationExchange::handleEnqueueEvent(const FieldTable* args, Deliverable
     std::string queueName = args->getAsString(REPLICATION_TARGET_QUEUE);
     Queue::shared_ptr queue = queues.find(queueName);
     if (queue) {
-        FieldTable& headers = msg.getMessage().getProperties<MessageProperties>()->getApplicationHeaders();
-        headers.erase(REPLICATION_TARGET_QUEUE);
-        headers.erase(REPLICATION_EVENT_SEQNO);
-        headers.erase(REPLICATION_EVENT_TYPE);
-        msg.deliverTo(queue);
-        QPID_LOG(debug, "Enqueued replicated message onto " << queueName);
-        if (mgmtExchange != 0) {
-            mgmtExchange->inc_msgRoutes();
-            mgmtExchange->inc_byteRoutes( msg.contentSize());
+
+        SequenceNumber seqno1(args->getAsInt(QUEUE_MESSAGE_POSITION));
+              
+        if (queue->getPosition() > seqno1) // test queue.pos < seqnumber
+        {
+            QPID_LOG(error, "Cannot enqueue replicated message. Destination Queue " << queueName << " ahead of source queue");
+            mgmtExchange->inc_msgDrops();
+            mgmtExchange->inc_byteDrops(msg.contentSize());
+        } else {
+            queue->setPosition(--seqno1);  // note that queue will ++ before enqueue.
+
+            FieldTable& headers = msg.getMessage().getProperties<MessageProperties>()->getApplicationHeaders();
+            headers.erase(REPLICATION_TARGET_QUEUE);
+            headers.erase(REPLICATION_EVENT_SEQNO);
+            headers.erase(REPLICATION_EVENT_TYPE);
+            msg.deliverTo(queue);
+            QPID_LOG(debug, "Enqueued replicated message onto " << queueName);
+            if (mgmtExchange != 0) {
+                mgmtExchange->inc_msgRoutes();
+                mgmtExchange->inc_byteRoutes( msg.contentSize());
+            }
         }
     } else {
         QPID_LOG(error, "Cannot enqueue replicated message. Queue " << queueName << " does not exist");
diff --git a/cpp/src/qpid/replication/constants.h b/cpp/src/qpid/replication/constants.h
index fb7085c570..c5ba7d3d6a 100644
--- a/cpp/src/qpid/replication/constants.h
+++ b/cpp/src/qpid/replication/constants.h
@@ -26,6 +26,7 @@ const std::string REPLICATION_EVENT_TYPE("qpid.replication.type");
 const std::string REPLICATION_EVENT_SEQNO("qpid.replication.seqno");
 const std::string REPLICATION_TARGET_QUEUE("qpid.replication.target_queue");
 const std::string DEQUEUED_MESSAGE_POSITION("qpid.replication.message");
+const std::string QUEUE_MESSAGE_POSITION("qpid.replication.queue.position");
 
 const int ENQUEUE(1);
 const int DEQUEUE(2);