diff options
author | Ted Ross <tross@apache.org> | 2011-07-12 18:29:22 +0000 |
---|---|---|
committer | Ted Ross <tross@apache.org> | 2011-07-12 18:29:22 +0000 |
commit | 2885957b73306b183b5889dd8bbfd8b58c61c521 (patch) | |
tree | bd7a55221e8ef11f2d95676f1fe2d860cd8cff9c | |
parent | fbbc3211abb59a99c5e9eb07dfd0f252a6416590 (diff) | |
download | qpid-python-2885957b73306b183b5889dd8bbfd8b58c61c521.tar.gz |
QPID-3352 - Federation bridge doesn't recover from session errors
Applied patch from Jason Dillaman
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1145706 13f79535-47bb-0310-9956-ffa450edef68
-rw-r--r-- | cpp/src/qpid/amqp_0_10/SessionHandler.cpp | 5 | ||||
-rw-r--r-- | cpp/src/qpid/broker/Bridge.cpp | 6 | ||||
-rw-r--r-- | cpp/src/qpid/broker/Bridge.h | 2 | ||||
-rw-r--r-- | cpp/src/qpid/broker/Link.cpp | 15 | ||||
-rwxr-xr-x | cpp/src/tests/federation.py | 57 | ||||
-rwxr-xr-x | cpp/src/tests/run_federation_tests | 2 |
6 files changed, 83 insertions, 4 deletions
diff --git a/cpp/src/qpid/amqp_0_10/SessionHandler.cpp b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp index 97281a8d8c..578598a146 100644 --- a/cpp/src/qpid/amqp_0_10/SessionHandler.cpp +++ b/cpp/src/qpid/amqp_0_10/SessionHandler.cpp @@ -188,9 +188,10 @@ void SessionHandler::detach(const std::string& name) { void SessionHandler::detached(const std::string& name, uint8_t code) { CHECK_NAME(name, "session.detached"); awaitingDetached = false; - if (code != session::DETACH_CODE_NORMAL) + if (code != session::DETACH_CODE_NORMAL) { + sendReady = receiveReady = false; channelException(convert(code), "session.detached from peer."); - else { + } else { handleDetach(); } } diff --git a/cpp/src/qpid/broker/Bridge.cpp b/cpp/src/qpid/broker/Bridge.cpp index 7fbbf4e2c4..c709606c17 100644 --- a/cpp/src/qpid/broker/Bridge.cpp +++ b/cpp/src/qpid/broker/Bridge.cpp @@ -164,6 +164,12 @@ void Bridge::destroy() listener(this); } +bool Bridge::isSessionReady() const +{ + SessionHandler& sessionHandler = conn->getChannel(id); + return sessionHandler.ready(); +} + void Bridge::setPersistenceId(uint64_t pId) const { persistenceId = pId; diff --git a/cpp/src/qpid/broker/Bridge.h b/cpp/src/qpid/broker/Bridge.h index a846254c57..8b4559a871 100644 --- a/cpp/src/qpid/broker/Bridge.h +++ b/cpp/src/qpid/broker/Bridge.h @@ -59,6 +59,8 @@ public: void destroy(); bool isDurable() { return args.i_durable; } + bool isSessionReady() const; + management::ManagementObject* GetManagementObject() const; management::Manageable::status_t ManagementMethod(uint32_t methodId, management::Args& args, diff --git a/cpp/src/qpid/broker/Link.cpp b/cpp/src/qpid/broker/Link.cpp index 9ab4379a69..8010bf43e7 100644 --- a/cpp/src/qpid/broker/Link.cpp +++ b/cpp/src/qpid/broker/Link.cpp @@ -248,6 +248,19 @@ void Link::ioThreadProcessing() return; QPID_LOG(debug, "Link::ioThreadProcessing()"); + // check for bridge session errors and recover + if (!active.empty()) { + Bridges::iterator removed = std::remove_if( + active.begin(), active.end(), !boost::bind(&Bridge::isSessionReady, _1)); + for (Bridges::iterator i = removed; i != active.end(); ++i) { + Bridge::shared_ptr bridge = *i; + bridge->closed(); + bridge->cancel(*connection); + created.push_back(bridge); + } + active.erase(removed, active.end()); + } + //process any pending creates and/or cancellations if (!created.empty()) { for (Bridges::iterator i = created.begin(); i != created.end(); ++i) { @@ -296,7 +309,7 @@ void Link::maintenanceVisit () } } } - else if (state == STATE_OPERATIONAL && (!created.empty() || !cancellations.empty()) && connection != 0) + else if (state == STATE_OPERATIONAL && (!active.empty() || !created.empty() || !cancellations.empty()) && connection != 0) connection->requestIOProcessing (boost::bind(&Link::ioThreadProcessing, this)); } diff --git a/cpp/src/tests/federation.py b/cpp/src/tests/federation.py index 49bdecdd95..0665712db3 100755 --- a/cpp/src/tests/federation.py +++ b/cpp/src/tests/federation.py @@ -268,6 +268,63 @@ class FederationTests(TestBase010): self.verify_cleanup() + def test_pull_from_queue_recovery(self): + session = self.session + + #setup queue on remote broker and add some messages + r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) + r_session = r_conn.session("test_pull_from_queue_recovery") + r_session.queue_declare(queue="my-bridge-queue", auto_delete=True) + for i in range(1, 6): + dp = r_session.delivery_properties(routing_key="my-bridge-queue") + r_session.message_transfer(message=Message(dp, "Message %d" % i)) + + #setup queue to receive messages from local broker + session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) + session.exchange_bind(queue="fed1", exchange="amq.fanout") + self.subscribe(queue="fed1", destination="f1") + queue = session.incoming("f1") + + self.startQmf() + qmf = self.qmf + broker = qmf.getObjects(_class="broker")[0] + result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") + self.assertEqual(result.status, 0) + + link = qmf.getObjects(_class="link")[0] + result = link.bridge(False, "my-bridge-queue", "amq.fanout", "my-key", "", "", True, False, False, 1) + self.assertEqual(result.status, 0) + + bridge = qmf.getObjects(_class="bridge")[0] + sleep(5) + + #recreate the remote bridge queue to invalidate the bridge session + r_session.queue_delete (queue="my-bridge-queue", if_empty=False, if_unused=False) + r_session.queue_declare(queue="my-bridge-queue", auto_delete=True) + + #add some more messages (i.e. after bridge was created) + for i in range(6, 11): + dp = r_session.delivery_properties(routing_key="my-bridge-queue") + r_session.message_transfer(message=Message(dp, "Message %d" % i)) + + for i in range(1, 11): + try: + msg = queue.get(timeout=5) + self.assertEqual("Message %d" % i, msg.body) + except Empty: + self.fail("Failed to find expected message containing 'Message %d'" % i) + try: + extra = queue.get(timeout=1) + self.fail("Got unexpected message in queue: " + extra.body) + except Empty: None + + result = bridge.close() + self.assertEqual(result.status, 0) + result = link.close() + self.assertEqual(result.status, 0) + + self.verify_cleanup() + def test_tracing_automatic(self): remoteUrl = "%s:%d" % (self.remote_host(), self.remote_port()) self.startQmf() diff --git a/cpp/src/tests/run_federation_tests b/cpp/src/tests/run_federation_tests index 590f74746e..14af4807ba 100755 --- a/cpp/src/tests/run_federation_tests +++ b/cpp/src/tests/run_federation_tests @@ -55,7 +55,7 @@ stop_brokers() { if test -d ${PYTHON_DIR} ; then start_brokers echo "Running federation tests using brokers on ports $LOCAL_PORT $REMOTE_PORT $REMOTE_B1 $REMOTE_B2" - $QPID_PYTHON_TEST -m federation $SKIPTESTS -b localhost:$LOCAL_PORT -Dremote-port=$REMOTE_PORT -Dextra-brokers="$REMOTE_B1 $REMOTE_B2" $@ + $QPID_PYTHON_TEST -m federation "$SKIPTESTS" -b localhost:$LOCAL_PORT -Dremote-port=$REMOTE_PORT -Dextra-brokers="$REMOTE_B1 $REMOTE_B2" $@ RETCODE=$? stop_brokers if test x$RETCODE != x0; then |