QPID-4710: Update message decoding logic to handle nested descriptors also

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

6 files changed, 53 insertions, 31 deletions

diff --git a/qpid/cpp/src/qpid/amqp/MessageReader.cpp b/qpid/cpp/src/qpid/amqp/MessageReader.cpp
index 70374f4cd6..ab90472067 100644
--- a/qpid/cpp/src/qpid/amqp/MessageReader.cpp
+++ b/qpid/cpp/src/qpid/amqp/MessageReader.cpp
@@ -54,6 +54,12 @@ const size_t GROUP_ID(10);
 const size_t GROUP_SEQUENCE(11);
 const size_t REPLY_TO_GROUP_ID(12);
 
+
+const Descriptor* nested(const Descriptor* d)
+{
+    if (d && d->nested) return d->nested.get();
+    else return 0;
+}
 }
 
 MessageReader::HeaderReader::HeaderReader(MessageReader& p) : parent(p), index(0) {}
@@ -269,7 +275,7 @@ bool MessageReader::onStartList(uint32_t count, const CharSequence& elements, co
             onAmqpSequence(raw);
             return false;
         } else if (descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
-            onAmqpValue(elements, qpid::amqp::typecodes::LIST_NAME);
+            onAmqpValue(elements, qpid::amqp::typecodes::LIST_NAME, nested(descriptor));
             return false;
         } else {
             QPID_LOG(warning, "Unexpected described list: " << *descriptor);
@@ -310,7 +316,7 @@ bool MessageReader::onStartMap(uint32_t count, const CharSequence& elements, con
             onApplicationProperties(elements, raw);
             return false;
         } else if (descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
-            onAmqpValue(elements, qpid::amqp::typecodes::MAP_NAME);
+            onAmqpValue(elements, qpid::amqp::typecodes::MAP_NAME, nested(descriptor));
             return false;
         } else {
             QPID_LOG(warning, "Unexpected described map: " << *descriptor);
@@ -337,7 +343,7 @@ void MessageReader::onBinary(const CharSequence& bytes, const Descriptor* descri
         } else if (descriptor->match(DATA_SYMBOL, DATA_CODE)) {
             onData(bytes);
         } else if (descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
-            onAmqpValue(bytes, qpid::amqp::typecodes::BINARY_NAME);
+            onAmqpValue(bytes, qpid::amqp::typecodes::BINARY_NAME, nested(descriptor));
         } else {
             QPID_LOG(warning, "Unexpected binary value with descriptor: " << *descriptor);
         }
@@ -353,7 +359,7 @@ void MessageReader::onNull(const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant v;
-            onAmqpValue(v);
+            onAmqpValue(v, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got null value with no descriptor.");
@@ -369,7 +375,7 @@ void MessageReader::onString(const CharSequence& v, const Descriptor* descriptor
         delegate->onString(v, descriptor);
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
-            onAmqpValue(v, qpid::amqp::typecodes::STRING_NAME);
+            onAmqpValue(v, qpid::amqp::typecodes::STRING_NAME, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got string value with no descriptor.");
@@ -385,7 +391,7 @@ void MessageReader::onSymbol(const CharSequence& v, const Descriptor* descriptor
         delegate->onSymbol(v, descriptor);
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
-            onAmqpValue(v, qpid::amqp::typecodes::SYMBOL_NAME);
+            onAmqpValue(v, qpid::amqp::typecodes::SYMBOL_NAME, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got symbol value with no descriptor.");
@@ -403,7 +409,7 @@ void MessageReader::onBoolean(bool v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got boolean value with no descriptor.");
@@ -421,7 +427,7 @@ void MessageReader::onUByte(uint8_t v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got ubyte value with no descriptor.");
@@ -439,7 +445,7 @@ void MessageReader::onUShort(uint16_t v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got ushort value with no descriptor.");
@@ -457,7 +463,7 @@ void MessageReader::onUInt(uint32_t v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got uint value with no descriptor.");
@@ -475,7 +481,7 @@ void MessageReader::onULong(uint64_t v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got ulong value with no descriptor.");
@@ -493,7 +499,7 @@ void MessageReader::onByte(int8_t v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got byte value with no descriptor.");
@@ -511,7 +517,7 @@ void MessageReader::onShort(int16_t v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got short value with no descriptor.");
@@ -529,7 +535,7 @@ void MessageReader::onInt(int32_t v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got int value with no descriptor.");
@@ -547,7 +553,7 @@ void MessageReader::onLong(int64_t v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got long value with no descriptor.");
@@ -565,7 +571,7 @@ void MessageReader::onFloat(float v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got float value with no descriptor.");
@@ -583,7 +589,7 @@ void MessageReader::onDouble(double v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got double value with no descriptor.");
@@ -600,7 +606,7 @@ void MessageReader::onUuid(const CharSequence& v, const Descriptor* descriptor)
         delegate->onUuid(v, descriptor);
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
-            onAmqpValue(v, qpid::amqp::typecodes::UUID_NAME);
+            onAmqpValue(v, qpid::amqp::typecodes::UUID_NAME, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got uuid value with no descriptor.");
@@ -618,7 +624,7 @@ void MessageReader::onTimestamp(int64_t v, const Descriptor* descriptor)
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             qpid::types::Variant body = v;
-            onAmqpValue(body);
+            onAmqpValue(body, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got timestamp value with no descriptor.");
@@ -636,7 +642,7 @@ bool MessageReader::onStartArray(uint32_t count, const CharSequence& raw, const
     } else {
         if (descriptor && descriptor->match(AMQP_VALUE_SYMBOL, AMQP_VALUE_CODE)) {
             //TODO: might be better to decode this here
-            onAmqpValue(raw, qpid::amqp::typecodes::ARRAY_NAME);
+            onAmqpValue(raw, qpid::amqp::typecodes::ARRAY_NAME, nested(descriptor));
         } else {
             if (!descriptor) {
                 QPID_LOG(warning, "Expected described type but got array with no descriptor.");
diff --git a/qpid/cpp/src/qpid/amqp/MessageReader.h b/qpid/cpp/src/qpid/amqp/MessageReader.h
index 4e2248070b..2fb588546f 100644
--- a/qpid/cpp/src/qpid/amqp/MessageReader.h
+++ b/qpid/cpp/src/qpid/amqp/MessageReader.h
@@ -100,8 +100,8 @@ class MessageReader : public Reader
 
     virtual void onData(const CharSequence&) = 0;
     virtual void onAmqpSequence(const CharSequence&) = 0;
-    virtual void onAmqpValue(const CharSequence&, const std::string& type) = 0;
-    virtual void onAmqpValue(const qpid::types::Variant&) = 0;
+    virtual void onAmqpValue(const CharSequence&, const std::string& type, const Descriptor*) = 0;
+    virtual void onAmqpValue(const qpid::types::Variant&, const Descriptor*) = 0;
 
     virtual void onFooter(const CharSequence& /*values*/, const CharSequence& /*full*/) = 0;
 
diff --git a/qpid/cpp/src/qpid/broker/amqp/Message.cpp b/qpid/cpp/src/qpid/broker/amqp/Message.cpp
index 3c8947d28e..54741e6436 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Message.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Message.cpp
@@ -217,7 +217,7 @@ std::string Message::getContent() const
     return std::string(body.data, body.size);
 }
 
-Message::Message(size_t size) : data(size)
+Message::Message(size_t size) : data(size), bodyDescriptor(0)
 {
     deliveryAnnotations.init();
     messageAnnotations.init();
@@ -329,7 +329,7 @@ void Message::onMessageAnnotations(const qpid::amqp::CharSequence&, const qpid::
 
 void Message::onData(const qpid::amqp::CharSequence& v) { body = v; }
 void Message::onAmqpSequence(const qpid::amqp::CharSequence& v) { body = v; bodyType = qpid::amqp::typecodes::LIST_NAME; }
-void Message::onAmqpValue(const qpid::amqp::CharSequence& v, const std::string& t)
+void Message::onAmqpValue(const qpid::amqp::CharSequence& v, const std::string& t, const qpid::amqp::Descriptor* d)
 {
     body = v;
     if (t == qpid::amqp::typecodes::STRING_NAME) {
@@ -341,8 +341,17 @@ void Message::onAmqpValue(const qpid::amqp::CharSequence& v, const std::string&
     } else {
         bodyType = t;
     }
+    if (d) {
+        bodyDescriptor = *d;
+    }
+}
+void Message::onAmqpValue(const qpid::types::Variant& v, const qpid::amqp::Descriptor* d)
+{
+    typedBody = v;
+    if (d) {
+        bodyDescriptor = *d;
+    }
 }
-void Message::onAmqpValue(const qpid::types::Variant& v) { typedBody = v; }
 
 void Message::onFooter(const qpid::amqp::CharSequence&, const qpid::amqp::CharSequence& v) { footer = v; }
 
@@ -372,6 +381,10 @@ qpid::types::Variant Message::getTypedBody() const
     }
 }
 
+const qpid::amqp::Descriptor& Message::getBodyDescriptor() const
+{
+    return bodyDescriptor;
+}
 
 //PersistableMessage interface:
 void Message::encode(framing::Buffer& buffer) const
diff --git a/qpid/cpp/src/qpid/broker/amqp/Message.h b/qpid/cpp/src/qpid/broker/amqp/Message.h
index 025e07c792..20310aa977 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Message.h
+++ b/qpid/cpp/src/qpid/broker/amqp/Message.h
@@ -23,6 +23,7 @@
  */
 #include "qpid/broker/Message.h"
 #include "qpid/amqp/CharSequence.h"
+#include "qpid/amqp/Descriptor.h"
 #include "qpid/amqp/MessageId.h"
 #include "qpid/amqp/MessageReader.h"
 #include <boost/optional.hpp>
@@ -67,6 +68,7 @@ class Message : public qpid::broker::Message::SharedStateImpl, private qpid::amq
     qpid::amqp::CharSequence getFooter() const;
     bool isTypedBody() const;
     qpid::types::Variant getTypedBody() const;
+    const qpid::amqp::Descriptor& getBodyDescriptor() const;
 
     Message(size_t size);
     char* getData();
@@ -114,6 +116,7 @@ class Message : public qpid::broker::Message::SharedStateImpl, private qpid::amq
     qpid::amqp::CharSequence body;
     qpid::types::Variant typedBody;
     std::string bodyType;
+    qpid::amqp::Descriptor bodyDescriptor;
 
     //footer:
     qpid::amqp::CharSequence footer;
@@ -147,8 +150,8 @@ class Message : public qpid::broker::Message::SharedStateImpl, private qpid::amq
 
     void onData(const qpid::amqp::CharSequence&);
     void onAmqpSequence(const qpid::amqp::CharSequence&);
-    void onAmqpValue(const qpid::amqp::CharSequence&, const std::string& type);
-    void onAmqpValue(const qpid::types::Variant&);
+    void onAmqpValue(const qpid::amqp::CharSequence&, const std::string& type, const qpid::amqp::Descriptor*);
+    void onAmqpValue(const qpid::types::Variant&, const qpid::amqp::Descriptor*);
 
     void onFooter(const qpid::amqp::CharSequence&, const qpid::amqp::CharSequence&);
 };
diff --git a/qpid/cpp/src/qpid/messaging/amqp/EncodedMessage.cpp b/qpid/cpp/src/qpid/messaging/amqp/EncodedMessage.cpp
index 77041d431a..cf60046245 100644
--- a/qpid/cpp/src/qpid/messaging/amqp/EncodedMessage.cpp
+++ b/qpid/cpp/src/qpid/messaging/amqp/EncodedMessage.cpp
@@ -345,7 +345,7 @@ void EncodedMessage::InitialScan::onAmqpSequence(const qpid::amqp::CharSequence&
     em.body = v;
     em.bodyType = qpid::amqp::typecodes::LIST_NAME;
 }
-void EncodedMessage::InitialScan::onAmqpValue(const qpid::amqp::CharSequence& v, const std::string& type)
+void EncodedMessage::InitialScan::onAmqpValue(const qpid::amqp::CharSequence& v, const std::string& type, const qpid::amqp::Descriptor*)
 {
     em.body = v;
     if (type == qpid::amqp::typecodes::STRING_NAME) {
@@ -356,7 +356,7 @@ void EncodedMessage::InitialScan::onAmqpValue(const qpid::amqp::CharSequence& v,
         em.bodyType = type;
     }
 }
-void EncodedMessage::InitialScan::onAmqpValue(const qpid::types::Variant& v)
+void EncodedMessage::InitialScan::onAmqpValue(const qpid::types::Variant& v, const qpid::amqp::Descriptor*)
 {
     em.content = v;
 }
diff --git a/qpid/cpp/src/qpid/messaging/amqp/EncodedMessage.h b/qpid/cpp/src/qpid/messaging/amqp/EncodedMessage.h
index 16a43aecea..241118386c 100644
--- a/qpid/cpp/src/qpid/messaging/amqp/EncodedMessage.h
+++ b/qpid/cpp/src/qpid/messaging/amqp/EncodedMessage.h
@@ -138,8 +138,8 @@ class EncodedMessage
 
         void onData(const qpid::amqp::CharSequence&);
         void onAmqpSequence(const qpid::amqp::CharSequence&);
-        void onAmqpValue(const qpid::amqp::CharSequence&, const std::string& type);
-        void onAmqpValue(const qpid::types::Variant&);
+        void onAmqpValue(const qpid::amqp::CharSequence&, const std::string& type, const qpid::amqp::Descriptor*);
+        void onAmqpValue(const qpid::types::Variant&, const qpid::amqp::Descriptor*);
 
         void onFooter(const qpid::amqp::CharSequence&, const qpid::amqp::CharSequence&);
       private: