QPID-4713: fix handling of reply to when converting from 1.0 to 0-10 format

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

4 files changed, 36 insertions, 10 deletions

diff --git a/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp b/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp
index 16f0cdc39f..91008d4075 100644
--- a/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp
+++ b/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp
@@ -114,7 +114,7 @@ qpid::sys::ConnectionCodec* ProtocolImpl::create(const qpid::framing::ProtocolVe
 
 boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> ProtocolImpl::translate(const qpid::broker::Message& m)
 {
-    qpid::broker::amqp::Translation t(m);
+    qpid::broker::amqp::Translation t(m, &broker);
     return t.getTransfer();
 }
 
diff --git a/cpp/src/qpid/broker/amqp/Translation.cpp b/cpp/src/qpid/broker/amqp/Translation.cpp
index 3191ff0a8a..846deb92f5 100644
--- a/cpp/src/qpid/broker/amqp/Translation.cpp
+++ b/cpp/src/qpid/broker/amqp/Translation.cpp
@@ -21,6 +21,7 @@
 #include "qpid/broker/amqp/Translation.h"
 #include "qpid/broker/amqp/Outgoing.h"
 #include "qpid/broker/amqp_0_10/MessageTransfer.h"
+#include "qpid/broker/Broker.h"
 #include "qpid/amqp/Decoder.h"
 #include "qpid/amqp/descriptors.h"
 #include "qpid/amqp/MessageEncoder.h"
@@ -38,10 +39,30 @@ namespace {
 const std::string EMPTY;
 const std::string FORWARD_SLASH("/");
 
+qpid::framing::ReplyTo translate(const std::string address, Broker* broker)
+{
+    size_t i = address.find(FORWARD_SLASH);
+    if (i == std::string::npos) {
+        //is it a queue or an exchange?
+        if (broker && broker->getQueues().find(address)) {
+            return qpid::framing::ReplyTo(EMPTY, address);
+        } else if (broker && broker->getExchanges().find(address)) {
+            return qpid::framing::ReplyTo(address, EMPTY);
+        } else {
+            return qpid::framing::ReplyTo();
+        }
+    } else {
+        return qpid::framing::ReplyTo(i > 0 ? address.substr(0, i) : EMPTY, (i+1) < address.size() ? address.substr(i+1) : EMPTY);
+    }
+}
+qpid::framing::ReplyTo translate(const qpid::amqp::CharSequence& address, Broker* broker)
+{
+    return translate(std::string(address.data, address.size), broker);
+}
 std::string translate(const qpid::framing::ReplyTo r)
 {
-    if (r.hasExchange()) {
-        if (r.hasRoutingKey()) return r.getExchange() + FORWARD_SLASH + r.getRoutingKey();
+    if (r.getExchange().size()) {
+        if (r.getRoutingKey().size()) return r.getExchange() + FORWARD_SLASH + r.getRoutingKey();
         else return r.getExchange();
     } else return r.getRoutingKey();
 }
@@ -115,7 +136,7 @@ class Properties_0_10 : public qpid::amqp::MessageEncoder::Properties
 };
 }
 
-Translation::Translation(const qpid::broker::Message& m) : original(m) {}
+Translation::Translation(const qpid::broker::Message& m, Broker* b) : original(m), broker(b) {}
 
 
 boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> Translation::getTransfer()
@@ -175,10 +196,7 @@ boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> Translation
                 props->setCorrelationId(boost::lexical_cast<std::string>(cid.value.ulong));
                 break;
             }
-            // TODO: ReplyTo - there is no way to reliably determine
-            // the type of the node from just its name, unless we
-            // query the brokers registries
-
+            if (message->getReplyTo()) props->setReplyTo(translate(message->getReplyTo(), broker));
             if (message->getContentType()) props->setContentType(translate(message->getContentType()));
             if (message->getContentEncoding()) props->setContentEncoding(translate(message->getContentEncoding()));
             props->setUserId(message->getUserId());
diff --git a/cpp/src/qpid/broker/amqp/Translation.h b/cpp/src/qpid/broker/amqp/Translation.h
index 7591f45b2a..47cb69d0ba 100644
--- a/cpp/src/qpid/broker/amqp/Translation.h
+++ b/cpp/src/qpid/broker/amqp/Translation.h
@@ -25,6 +25,7 @@
 
 namespace qpid {
 namespace broker {
+class Broker;
 class Message;
 namespace amqp_0_10 {
 class MessageTransfer;
@@ -38,7 +39,7 @@ class OutgoingFromQueue;
 class Translation
 {
   public:
-    Translation(const qpid::broker::Message& message);
+    Translation(const qpid::broker::Message& message, Broker* broker = 0);
 
     /**
      * @returns a pointer to an AMQP 0-10 message transfer suitable
@@ -52,6 +53,7 @@ class Translation
     void write(OutgoingFromQueue&);
   private:
     const qpid::broker::Message& original;
+    Broker* broker;
 };
 }}} // namespace qpid::broker::amqp
 
diff --git a/cpp/src/qpid/messaging/amqp/SenderContext.cpp b/cpp/src/qpid/messaging/amqp/SenderContext.cpp
index 0b3de7e680..aeb89d145c 100644
--- a/cpp/src/qpid/messaging/amqp/SenderContext.cpp
+++ b/cpp/src/qpid/messaging/amqp/SenderContext.cpp
@@ -134,6 +134,7 @@ class HeaderAdapter : public qpid::amqp::MessageEncoder::Header
     const qpid::messaging::MessageImpl& msg;
 };
 const std::string EMPTY;
+const std::string FORWARD_SLASH("/");
 
 class PropertiesAdapter : public qpid::amqp::MessageEncoder::Properties
 {
@@ -185,7 +186,12 @@ class PropertiesAdapter : public qpid::amqp::MessageEncoder::Properties
 
     std::string getReplyTo() const
     {
-        return msg.getReplyTo().str();
+        Address a = msg.getReplyTo();
+        if (a.getSubject().size()) {
+            return a.getName() + FORWARD_SLASH + a.getSubject();
+        } else {
+            return a.getName();
+        }
     }
 
     bool hasCorrelationId() const