diff options
Diffstat (limited to 'cpp')
| -rw-r--r-- | cpp/src/qpid/replication/ReplicationExchange.cpp | 33 | ||||
| -rwxr-xr-x | cpp/src/tests/replication_test | 6 |
2 files changed, 26 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"); } } diff --git a/cpp/src/tests/replication_test b/cpp/src/tests/replication_test index 9b6e5cfb29..605ee1376c 100755 --- a/cpp/src/tests/replication_test +++ b/cpp/src/tests/replication_test @@ -55,10 +55,12 @@ if test -d ${PYTHON_DIR} && test -e ../.libs/replicating_listener.so && test -e $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_A" add queue queue-a --generate-queue-events 2 $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_A" add queue queue-b --generate-queue-events 2 $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_A" add queue queue-c --generate-queue-events 1 + $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_A" add queue queue-d --generate-queue-events 2 $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_B" add queue queue-a $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_B" add queue queue-b $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_B" add queue queue-c + #queue-d deliberately not declared on DR; this error case should be handled #publish and consume from test queus on broker A: for i in `seq 1 10`; do echo Message $i for A >> queue-a-input.repl; done @@ -68,10 +70,12 @@ if test -d ${PYTHON_DIR} && test -e ../.libs/replicating_listener.so && test -e ./sender --port $BROKER_A --routing-key queue-a --send-eos 1 < queue-a-input.repl ./sender --port $BROKER_A --routing-key queue-b --send-eos 1 < queue-b-input.repl ./sender --port $BROKER_A --routing-key queue-c --send-eos 1 < queue-c-input.repl + echo dummy | ./sender --port $BROKER_A --routing-key queue-d --send-eos 1 ./receiver --port $BROKER_A --queue queue-a --messages 5 > /dev/null ./receiver --port $BROKER_A --queue queue-b --messages 10 > /dev/null ./receiver --port $BROKER_A --queue queue-c --messages 10 > /dev/null + ./receiver --port $BROKER_A --queue queue-d > /dev/null #shutdown broker A then check that broker Bs versions of the queues are as expected ../qpidd -q --port $BROKER_A @@ -90,6 +94,8 @@ if test -d ${PYTHON_DIR} && test -e ../.libs/replicating_listener.so && test -e diff queue-b-backup.repl queue-b-expected.repl || FAIL=1 diff queue-c-backup.repl queue-c-input.repl || FAIL=1 + grep 'queue-d does not exist' replication-dest.log > /dev/null || echo "WARNING: Expected error to be logged!" + if [[ $FAIL ]]; then echo replication test failed: expectations not met! exit 1 |