summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorTed Ross <tross@apache.org>2011-07-12 18:29:22 +0000
committerTed Ross <tross@apache.org>2011-07-12 18:29:22 +0000
commit2885957b73306b183b5889dd8bbfd8b58c61c521 (patch)
treebd7a55221e8ef11f2d95676f1fe2d860cd8cff9c
parentfbbc3211abb59a99c5e9eb07dfd0f252a6416590 (diff)
downloadqpid-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.cpp5
-rw-r--r--cpp/src/qpid/broker/Bridge.cpp6
-rw-r--r--cpp/src/qpid/broker/Bridge.h2
-rw-r--r--cpp/src/qpid/broker/Link.cpp15
-rwxr-xr-xcpp/src/tests/federation.py57
-rwxr-xr-xcpp/src/tests/run_federation_tests2
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