QPID-1567: added 'exactly-once' guarantee to asynchronous replication of queue state

* altered replication protocol to detect and eliminate duplicates
* added support for acknowledged transfer over inter-broker bridges
* added option to qpid-route to control this



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

1 files changed, 98 insertions, 0 deletions

diff --git a/qpid/cpp/src/tests/reliable_replication_test b/qpid/cpp/src/tests/reliable_replication_test
new file mode 100755
index 0000000000..1a2fa917b3
--- /dev/null
+++ b/qpid/cpp/src/tests/reliable_replication_test
@@ -0,0 +1,98 @@
+#!/bin/sh
+
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+# 
+#   http://www.apache.org/licenses/LICENSE-2.0
+# 
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+# Test reliability of the replication feature in the face of link
+# failures:
+MY_DIR=`dirname \`which $0\``
+PYTHON_DIR=${MY_DIR}/../../../python
+
+trap stop_brokers EXIT
+
+stop_brokers() {
+    if [[ $BROKER_A ]] ; then
+        ../qpidd -q --port $BROKER_A
+        unset BROKER_A
+    fi      
+    if [[ $BROKER_B ]] ; then
+        ../qpidd -q --port $BROKER_B
+        unset BROKER_B
+    fi
+}
+
+setup() {
+    rm -f replication-source.log replication-dest.log
+    ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no --load-module ../.libs/replicating_listener.so --replication-queue replication --create-replication-queue true  --log-enable trace+ --log-to-file replication-source.log --log-to-stderr 0 > qpidd.port
+    BROKER_A=`cat qpidd.port`
+
+    ../qpidd --daemon --port 0 --no-data-dir --no-module-dir --auth no --load-module ../.libs/replication_exchange.so  --log-enable info+ --log-to-file replication-dest.log --log-to-stderr 0 > qpidd.port
+    BROKER_B=`cat qpidd.port`
+
+    #../qpidd --port 5555 --no-data-dir --no-module-dir --auth no --load-module ../.libs/replicating_listener.so --replication-queue replication --create-replication-queue true  --log-enable trace+ --log-to-file replication-source.log --log-to-stderr 0 &
+    #BROKER_A=5555
+
+    #../qpidd --port 6666 --no-data-dir --no-module-dir --auth no --load-module ../.libs/replication_exchange.so  --log-enable info+ --log-to-file replication-dest.log --log-to-stderr 0 &
+    #BROKER_B=6666
+    echo "Testing replication from port $BROKER_A to port $BROKER_B"
+    export PYTHONPATH=$PYTHON_DIR
+
+    $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_B" add exchange replication replication
+    $PYTHON_DIR/commands/qpid-route --ack 500 queue add "localhost:$BROKER_B" "localhost:$BROKER_A" replication replication
+
+    #create test queue (only replicate enqueues for this test):
+    $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_A" add queue queue-a --generate-queue-events 1
+    $PYTHON_DIR/commands/qpid-config -a "localhost:$BROKER_B" add queue queue-a
+}
+
+send() {
+    ./sender  --port $BROKER_A --routing-key queue-a --send-eos 1 < replicated.expected
+}
+
+receive() {
+    rm -f replicated.actual
+    ./receiver --port $BROKER_B --queue queue-a > replicated.actual
+}
+
+bounce_link() {
+    echo "Destroying link..."
+    $PYTHON_DIR/commands/qpid-route link del "localhost:$BROKER_B" "localhost:$BROKER_A"
+    echo "Link destroyed; recreating route..."
+    sleep 2
+    $PYTHON_DIR/commands/qpid-route --ack 500 queue add "localhost:$BROKER_B" "localhost:$BROKER_A" replication replication
+    echo "Route re-established"
+}
+
+if test -d ${PYTHON_DIR} && test -e ../.libs/replicating_listener.so && test -e ../.libs/replication_exchange.so ;  then
+    setup
+    for i in `seq 1 100000`; do echo Message $i; done > replicated.expected
+    send &
+    receive &
+    for i in `seq 1 5`; do sleep 10; bounce_link; done;
+    wait
+    #check that received list is identical to sent list
+    diff replicated.actual replicated.expected || FAIL=1
+    if [[ $FAIL ]]; then
+        echo reliable replication test failed: expectations not met!
+    else 
+        echo replication reliable in the face of link failures
+        rm -f replication.actual replication.expected replication-source.log replication-dest.log
+    fi
+fi
+