Author: gsim
Date: Fri Jul 6 18:01:07 2012
New Revision: 1358321
URL: http://svn.apache.org/viewvc?rev=1358321&view=rev
Log:
QPID-4117: honour alternate-exchange option specified within x-declare for a
link
Modified:
qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp?rev=1358321&r1=1358320&r2=1358321&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
(original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp Fri Jul
6 18:01:07 2012
@@ -241,6 +241,7 @@ class Subscription : public Exchange, pu
const std::string actualType;
const bool exclusiveQueue;
const bool exclusiveSubscription;
+ const std::string alternateExchange;
FieldTable queueOptions;
FieldTable subscriptionOptions;
Bindings bindings;
@@ -507,7 +508,8 @@ Subscription::Subscription(const Address
durable(Opt(address)/LINK/DURABLE),
actualType(type.empty() ? (specifiedType.empty() ? TOPIC_EXCHANGE :
specifiedType) : type),
exclusiveQueue((Opt(address)/LINK/X_DECLARE/EXCLUSIVE).asBool(true)),
-
exclusiveSubscription((Opt(address)/LINK/X_SUBSCRIBE/EXCLUSIVE).asBool(exclusiveQueue))
+
exclusiveSubscription((Opt(address)/LINK/X_SUBSCRIBE/EXCLUSIVE).asBool(exclusiveQueue)),
+ alternateExchange((Opt(address)/LINK/X_DECLARE/ALTERNATE_EXCHANGE).str())
{
(Opt(address)/LINK/X_DECLARE/ARGUMENTS).collect(queueOptions);
(Opt(address)/LINK/X_SUBSCRIBE/ARGUMENTS).collect(subscriptionOptions);
@@ -568,7 +570,9 @@ void Subscription::subscribe(qpid::clien
//create subscription queue:
session.queueDeclare(arg::queue=queue, arg::exclusive=exclusiveQueue,
- arg::autoDelete=!reliable, arg::durable=durable,
arg::arguments=queueOptions);
+ arg::autoDelete=!reliable, arg::durable=durable,
+ arg::alternateExchange=alternateExchange,
+ arg::arguments=queueOptions);
//'default' binding:
bindings.bind(session);
//any explicit bindings:
Modified: qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp?rev=1358321&r1=1358320&r2=1358321&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp Fri Jul 6 18:01:07
2012
@@ -1146,6 +1146,24 @@ QPID_AUTO_TEST_CASE(testLargeRoutingKey)
BOOST_CHECK_THROW(fix.session.createReceiver(address),
qpid::messaging::MessagingException);
}
+QPID_AUTO_TEST_CASE(testAlternateExchangeInLinkDeclare)
+{
+ MessagingFixture fix;
+ Sender s = fix.session.createSender("amq.direct/key");
+ Receiver r1 =
fix.session.createReceiver("amq.direct/key;{link:{x-declare:{alternate-exchange:'amq.fanout'}}}");
+ Receiver r2 = fix.session.createReceiver("amq.fanout");
+
+ for (uint i = 0; i < 10; ++i) {
+ s.send(Message((boost::format("Message_%1%") % (i+1)).str()), true);
+ }
+ r1.close();//orphans all messages in subscription queue, which should then
be routed through alternate exchange
+ for (uint i = 0; i < 10; ++i) {
+ Message received;
+ BOOST_CHECK(r2.fetch(received, Duration::SECOND));
+ BOOST_CHECK_EQUAL(received.getContent(), (boost::format("Message_%1%")
% (i+1)).str());
+ }
+}
+
QPID_AUTO_TEST_SUITE_END()
}} // namespace qpid::tests
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
