QPID-3858: Updated branch - merged from trunk r.1333987

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

3 files changed, 12 insertions, 5 deletions

diff --git a/cpp/src/qpid/replication/ReplicatingEventListener.cpp b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
index 0ced4d9161..9284bda388 100644
--- a/cpp/src/qpid/replication/ReplicatingEventListener.cpp
+++ b/cpp/src/qpid/replication/ReplicatingEventListener.cpp
@@ -80,7 +80,7 @@ void ReplicatingEventListener::route(boost::intrusive_ptr<qpid::broker::Message>
     try {
         if (exchange) {
             DeliverableMessage deliverable(msg);
-            exchange->route(deliverable, msg->getRoutingKey(), msg->getApplicationHeaders());
+            exchange->route(deliverable);
         } else if (queue) {
             queue->deliver(msg);
         } else {
@@ -131,7 +131,7 @@ boost::intrusive_ptr<Message> ReplicatingEventListener::cloneMessage(Queue& queu
     //cloned body:
     AMQFrame header(*original->getFrames().getHeaders());
     header.setBof(false);
-    header.setEof(!original->getFrames().getContentSize());//if there is any content then the header is not the end of the frameset
+    header.setEof(!original->getFrames().hasContent());//if there are any content frames then the header is not the end of the frameset
     header.setBos(true);
     header.setEos(true);
     handler.handle(header);
diff --git a/cpp/src/qpid/replication/ReplicationExchange.cpp b/cpp/src/qpid/replication/ReplicationExchange.cpp
index 89a2bf516d..66f4f14d0c 100644
--- a/cpp/src/qpid/replication/ReplicationExchange.cpp
+++ b/cpp/src/qpid/replication/ReplicationExchange.cpp
@@ -50,8 +50,9 @@ ReplicationExchange::ReplicationExchange(const std::string& name, bool durable,
 
 std::string ReplicationExchange::getType() const { return typeName; }            
 
-void ReplicationExchange::route(Deliverable& msg, const std::string& /*routingKey*/, const FieldTable* args)
+void ReplicationExchange::route(Deliverable& msg)
 {
+    const FieldTable* args = msg.getMessage().getApplicationHeaders();
     if (mgmtExchange != 0) {
         mgmtExchange->inc_msgReceives();
         mgmtExchange->inc_byteReceives(msg.contentSize());
@@ -59,7 +60,13 @@ void ReplicationExchange::route(Deliverable& msg, const std::string& /*routingKe
     if (args) {
         int eventType = args->getAsInt(REPLICATION_EVENT_TYPE);
         if (eventType) {
-            if (isDuplicate(args)) return;
+            if (isDuplicate(args)) {
+                if (mgmtExchange != 0) {
+                    mgmtExchange->inc_msgDrops();
+                    mgmtExchange->inc_byteDrops(msg.contentSize());
+                }
+	        return;
+	    }
             switch (eventType) {
               case ENQUEUE:
                 handleEnqueueEvent(args, msg);
diff --git a/cpp/src/qpid/replication/ReplicationExchange.h b/cpp/src/qpid/replication/ReplicationExchange.h
index 4b34e0df13..ff0a98c48e 100644
--- a/cpp/src/qpid/replication/ReplicationExchange.h
+++ b/cpp/src/qpid/replication/ReplicationExchange.h
@@ -52,7 +52,7 @@ class ReplicationExchange : public qpid::broker::Exchange
 
     std::string getType() const;
 
-    void route(qpid::broker::Deliverable& msg, const std::string& routingKey, const qpid::framing::FieldTable* args);
+    void route(qpid::broker::Deliverable& msg);
         
     bool bind(boost::shared_ptr<broker::Queue> queue, const std::string& routingKey, const qpid::framing::FieldTable* args);
     bool unbind(boost::shared_ptr<broker::Queue> queue, const std::string& routingKey, const qpid::framing::FieldTable* args);