QPID-4117: honour alternate-exchange option specified within x-declare for a link

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

2 files changed, 24 insertions, 2 deletions

diff --git a/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp b/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
index a8f4fb5237..2627c178f9 100644
--- a/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
+++ b/qpid/cpp/src/qpid/client/amqp0_10/AddressResolution.cpp
@@ -241,6 +241,7 @@ class Subscription : public Exchange, public MessageSource
     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& address, const std::string& type)
       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::client::AsyncSession& session, const std::str
 
     //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:
diff --git a/qpid/cpp/src/tests/MessagingSessionTests.cpp b/qpid/cpp/src/tests/MessagingSessionTests.cpp
index 968d55fd45..c8ee3aa401 100644
--- a/qpid/cpp/src/tests/MessagingSessionTests.cpp
+++ b/qpid/cpp/src/tests/MessagingSessionTests.cpp
@@ -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