From 973ed9150310e9ba55ae69979d318f4bb256d0b2 Mon Sep 17 00:00:00 2001 From: "Carl C. Trieloff" Date: Thu, 30 Jul 2009 20:30:18 +0000 Subject: 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 --- cpp/src/qpid/replication/ReplicationExchange.cpp | 30 +++++++++++++++++------- 1 file changed, 21 insertions(+), 9 deletions(-) (limited to 'cpp/src/qpid/replication/ReplicationExchange.cpp') 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()->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()->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"); -- cgit v1.2.1