Author: aconway
Date: Wed Mar 2 21:06:51 2011
New Revision: 1076375
URL: http://svn.apache.org/viewvc?rev=1076375&view=rev
Log:
QPID-3105: Alternate-Exchange configuration not communicated between nodes in a
cluster.
Modified:
qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
Modified: qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp?rev=1076375&r1=1076374&r2=1076375&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/cluster/Cluster.cpp Wed Mar 2 21:06:51 2011
@@ -943,6 +943,10 @@ void Cluster::checkUpdateIn(Lock& l) {
mAgent->suppress(false); // Enable management output.
mAgent->clusterUpdate();
}
+ // Restore alternate exchange settings on exchanges.
+ broker.getExchanges().eachExchange(
+ boost::bind(&broker::Exchange::recoveryComplete, _1,
+ boost::ref(broker.getExchanges())));
enableClusterSafe(); // Enable cluster-safe assertions
deliverEventQueue.start();
}
Modified: qpid/trunk/qpid/cpp/src/tests/cluster_tests.py
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/cluster_tests.py?rev=1076375&r1=1076374&r2=1076375&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/cluster_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/cluster_tests.py Wed Mar 2 21:06:51 2011
@@ -22,7 +22,7 @@ import os, signal, sys, time, imp, re, s
from qpid import datatypes, messaging
from brokertest import *
from qpid.harness import Skipped
-from qpid.messaging import Message, Empty
+from qpid.messaging import Message, Empty, Disposition, REJECTED
from threading import Thread, Lock, Condition
from logging import getLogger
from itertools import chain
@@ -441,6 +441,33 @@ acl allow all all
return cluster[1]
self.queue_flowlimit_test(Brokers())
+ def test_alternate_exchange_update(self):
+ """Verify that alternate-exchange on exchanges and queues is
propagated to new members of a cluster. """
+ cluster = self.cluster(1)
+ s0 = cluster[0].connect().session()
+ # create alt queue bound to amq.fanout exchange, will be destination
for alternate exchanges
+ self.evaluate_address(s0,
"alt;{create:always,node:{x-bindings:[{exchange:'amq.fanout',queue:alt}]}}")
+ # create direct exchange ex with alternate-exchange amq.fanout and no
queues bound
+ self.evaluate_address(s0, "ex;{create:always,node:{type:topic,
x-declare:{type:'direct', alternate-exchange:'amq.fanout'}}}")
+ # create queue q with alternate-exchange amq.fanout
+ self.evaluate_address(s0, "q;{create:always,node:{type:queue,
x-declare:{alternate-exchange:'amq.fanout'}}}")
+
+ def verify(broker):
+ s = broker.connect().session()
+ # Verify unmatched message goes to ex's alternate.
+ s.sender("ex").send("foo")
+ self.assertEqual("foo", s.receiver("alt").fetch(timeout=0).content)
+ # Verify rejected message goes to q's alternate.
+ s.sender("q").send("bar")
+ msg = s.receiver("q").fetch(timeout=0)
+ self.assertEqual("bar", msg.content)
+ s.acknowledge(msg, Disposition(REJECTED)) # Reject the message
+ self.assertEqual("bar", s.receiver("alt").fetch(timeout=0).content)
+
+ verify(cluster[0])
+ cluster.start()
+ verify(cluster[1])
+
class LongTests(BrokerTest):
"""Tests that can run for a long time if -DDURATION=<minutes> is set"""
def duration(self):
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
