QPID-4625: Extract amqp 1.0 application-properties from messages

- A working implementaion to qpid::amqp::Message::processProperties()
  so that qpid::broker::Message::getProperty() will work.
* This is used by the selector implementation.

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

6 files changed, 81 insertions, 5 deletions

diff --git a/qpid/cpp/src/qpid/broker/HeadersExchange.cpp b/qpid/cpp/src/qpid/broker/HeadersExchange.cpp
index 37864a02f6..2d7c33abed 100644
--- a/qpid/cpp/src/qpid/broker/HeadersExchange.cpp
+++ b/qpid/cpp/src/qpid/broker/HeadersExchange.cpp
@@ -19,6 +19,8 @@
  *
  */
 #include "qpid/broker/HeadersExchange.h"
+
+#include "qpid/amqp/CharSequence.h"
 #include "qpid/broker/MapHandler.h"
 #include "qpid/framing/FieldValue.h"
 #include "qpid/framing/reply_exceptions.h"
diff --git a/qpid/cpp/src/qpid/broker/MapHandler.h b/qpid/cpp/src/qpid/broker/MapHandler.h
index 8f1acc76e1..15e3a95252 100644
--- a/qpid/cpp/src/qpid/broker/MapHandler.h
+++ b/qpid/cpp/src/qpid/broker/MapHandler.h
@@ -24,6 +24,11 @@
 #include "qpid/sys/IntegerTypes.h"
 
 namespace qpid {
+
+namespace amqp {
+struct CharSequence;
+}
+
 namespace broker {
 
 /**
@@ -32,10 +37,7 @@ namespace broker {
 class MapHandler
 {
   public:
-    typedef struct {
-        const char* data;
-        size_t size;
-    } CharSequence;
+    typedef qpid::amqp::CharSequence CharSequence;
 
     virtual ~MapHandler() {}
     virtual void handleVoid(const CharSequence& key) = 0;
diff --git a/qpid/cpp/src/qpid/broker/Message.cpp b/qpid/cpp/src/qpid/broker/Message.cpp
index d7072133c8..6903d27ae7 100644
--- a/qpid/cpp/src/qpid/broker/Message.cpp
+++ b/qpid/cpp/src/qpid/broker/Message.cpp
@@ -20,6 +20,8 @@
  */
 
 #include "qpid/broker/Message.h"
+
+#include "qpid/amqp/CharSequence.h"
 #include "qpid/broker/MapHandler.h"
 #include "qpid/StringUtils.h"
 #include "qpid/log/Statement.h"
diff --git a/qpid/cpp/src/qpid/broker/amqp/Message.cpp b/qpid/cpp/src/qpid/broker/amqp/Message.cpp
index a4c346e131..2c39c517e1 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Message.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Message.cpp
@@ -21,7 +21,9 @@
 #include "Message.h"
 #include "qpid/amqp/Decoder.h"
 #include "qpid/amqp/descriptors.h"
+#include "qpid/amqp/Reader.h"
 #include "qpid/amqp/MessageEncoder.h"
+#include "qpid/broker/MapHandler.h"
 #include "qpid/log/Statement.h"
 #include "qpid/framing/Buffer.h"
 #include <string.h>
@@ -30,6 +32,11 @@ namespace qpid {
 namespace broker {
 namespace amqp {
 
+using qpid::amqp::CharSequence;
+using qpid::amqp::Constructor;
+using qpid::amqp::Descriptor;
+using qpid::amqp::Reader;
+
 namespace {
 std::string empty;
 }
@@ -69,7 +76,67 @@ uint8_t Message::getPriority() const
 
 std::string Message::getPropertyAsString(const std::string& /*key*/) const { return empty; }
 std::string Message::getAnnotationAsString(const std::string& /*key*/) const { return empty; }
-void Message::processProperties(MapHandler&) const {}
+
+namespace {
+    class PropertyAdapter : public Reader {
+        MapHandler& handler;
+        CharSequence key;
+        enum {
+            KEY,
+            VALUE
+        } state;
+
+        void checkValue() {
+            if ( state==VALUE ) state = KEY;
+            else {
+                // TODO: Would throwing an exception make more sense here?
+                QPID_LOG(error, "Received non string property key");
+                key = CharSequence();
+                state = KEY;
+            }
+        }
+
+        void onNull(const Descriptor*) {checkValue(); handler.handleVoid(key);}
+        void onBoolean(bool b, const Descriptor*) {checkValue(); handler.handleBool(key, b);}
+        void onUByte(uint8_t i, const Descriptor*) {checkValue(); handler.handleUint8(key, i);}
+        void onUShort(uint16_t i, const Descriptor*) {checkValue(); handler.handleUint16(key, i);}
+        void onUInt(uint32_t i, const Descriptor*) {checkValue(); handler.handleUint32(key, i);}
+        void onULong(uint64_t i, const Descriptor*) {checkValue(); handler.handleUint64(key, i);}
+        void onByte(int8_t i, const Descriptor*) {checkValue(); handler.handleInt8(key, i);}
+        void onShort(int16_t i, const Descriptor*) {checkValue(); handler.handleInt16(key, i);}
+        void onInt(int32_t i, const Descriptor*) {checkValue(); handler.handleInt32(key, i);}
+        void onLong(int64_t i, const Descriptor*) {checkValue(); handler.handleInt64(key, i);}
+        void onFloat(float x, const Descriptor*) {checkValue(); handler.handleFloat(key, x);}
+        void onDouble(double x, const Descriptor*) {checkValue(); handler.handleDouble(key, x);}
+        void onTimestamp(int64_t i, const Descriptor*) {checkValue(); handler.handleInt64(key, i);}
+
+        void onString(const CharSequence& s, const Descriptor*) {
+            if ( state==KEY ) {
+                state = VALUE;
+                key = s;
+            } else {
+                state = KEY;
+                handler.handleString(key, s, CharSequence());
+            }
+        }
+
+        bool onStartList(uint32_t, const CharSequence&, const Descriptor*) { return false; }
+        bool onStartMap(uint32_t, const CharSequence&, const Descriptor*) { return false; }
+        bool onStartArray(uint32_t, const CharSequence&, const Constructor&, const Descriptor*) { return false; }
+
+    public:
+        PropertyAdapter(MapHandler& mh) :
+            handler(mh),
+            state(KEY)
+         {}
+    };
+}
+
+void Message::processProperties(MapHandler& mh) const {
+    qpid::amqp::Decoder d(applicationProperties.data, applicationProperties.size);
+    PropertyAdapter mha(mh);
+    d.read(mha);
+}
 
 //getContentSize() is primarily used in stats about the number of
 //bytes enqueued/dequeued etc, not sure whether this is the right name
diff --git a/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp b/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp
index d120d61eb1..b9c509c015 100644
--- a/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp
@@ -19,6 +19,8 @@
  *
  */
 #include "MessageTransfer.h"
+
+#include "qpid/amqp/CharSequence.h"
 #include "qpid/amqp_0_10/Codecs.h"
 #include "qpid/broker/MapHandler.h"
 #include "qpid/broker/Message.h"
diff --git a/qpid/cpp/src/qpid/xml/XmlExchange.cpp b/qpid/cpp/src/qpid/xml/XmlExchange.cpp
index a1b0c052d3..1f6bf6de05 100644
--- a/qpid/cpp/src/qpid/xml/XmlExchange.cpp
+++ b/qpid/cpp/src/qpid/xml/XmlExchange.cpp
@@ -23,6 +23,7 @@
 
 #include "qpid/xml/XmlExchange.h"
 
+#include "qpid/amqp/CharSequence.h"
 #include "qpid/broker/DeliverableMessage.h"
 
 #include "qpid/log/Statement.h"