QPID-1717: Check that target queue for replication events exists before attempting to process it.

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

1 files changed, 20 insertions, 13 deletions

diff --git a/cpp/src/qpid/replication/ReplicationExchange.cpp b/cpp/src/qpid/replication/ReplicationExchange.cpp
index 79a0b0861c..c35c6c2cd5 100644
--- a/cpp/src/qpid/replication/ReplicationExchange.cpp
+++ b/cpp/src/qpid/replication/ReplicationExchange.cpp
@@ -68,26 +68,33 @@ void ReplicationExchange::handleEnqueueEvent(const FieldTable* args, Deliverable
 {
     std::string queueName = args->getAsString(REPLICATION_TARGET_QUEUE);
     Queue::shared_ptr queue = queues.find(queueName);
-    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 (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);
+    } else {
+        QPID_LOG(error, "Cannot enqueue replicated message. Queue " << queueName << " does not exist");
+    }
 }
 
 void ReplicationExchange::handleDequeueEvent(const FieldTable* args)
 {
     std::string queueName = args->getAsString(REPLICATION_TARGET_QUEUE);
     Queue::shared_ptr queue = queues.find(queueName);
-    SequenceNumber position(args->getAsInt(DEQUEUED_MESSAGE_POSITION));
-    
-    QueuedMessage dequeued;
-    if (queue->acquireMessageAt(position, dequeued)) {
-        queue->dequeue(0, dequeued);
-        QPID_LOG(debug, "Processed replicated 'dequeue' event from " << queueName << " at position " << position);
+    if (queue) {
+        SequenceNumber position(args->getAsInt(DEQUEUED_MESSAGE_POSITION));        
+        QueuedMessage dequeued;
+        if (queue->acquireMessageAt(position, dequeued)) {
+            queue->dequeue(0, dequeued);
+            QPID_LOG(debug, "Processed replicated 'dequeue' event from " << queueName << " at position " << position);
+        } else {
+            QPID_LOG(warning, "Could not acquire message " << position << " from " << queueName);
+        }
     } else {
-        QPID_LOG(warning, "Could not acquire message " << position << " from " << queueName);
+        QPID_LOG(error, "Cannot process replicated 'dequeue' event. Queue " << queueName << " does not exist");
     }
 }