Currently we add a number of custom properties to the properties section of the AMQP message (e.g. MsgType, Inbox Name). The properties section of the message is used for a defined set of standard properties of the AMQP spec (message-id, user-id etc), and is not meant to be extensible for custom properties. The Application Properties section of the AMQP message is the section for adding custom properties, in the form of a key (string), value map.
This change moves the following information from properties to application properties:
* Message Type
* Inbox Name
* Reply To
* Target Subject
Tested on
* Mac OS X 10.9.2 x86_64
* Ubuntu 12.04 x86_64
Tested for:
* Pub / Sub
* Pub / Inbox
* CaptureReplay / MamaListen
* Unit Tests
Signed-off-by: Phil Preston <philippreston@...>
---
mama/c_cpp/src/c/bridge/qpid/codec.c | 82 +++++++++++++++++++++++++-------
mama/c_cpp/src/c/bridge/qpid/publisher.c | 9 ++--
mama/c_cpp/src/c/bridge/qpid/qpiddefs.h | 5 ++
mama/c_cpp/src/c/bridge/qpid/transport.c | 19 ++++++--
4 files changed, 89 insertions(+), 26 deletions(-)
diff --git a/mama/c_cpp/src/c/bridge/qpid/codec.c b/mama/c_cpp/src/c/bridge/qpid/codec.c
index 936ade0..27379c5 100644
--- a/mama/c_cpp/src/c/bridge/qpid/codec.c
+++ b/mama/c_cpp/src/c/bridge/qpid/codec.c
@@ -104,8 +104,8 @@ qpidBridgeMsgCodec_pack (msgBridge bridgeMessage,
/* Ensure position is at the start */
pn_data_rewind (properties);
- /* Main container for meta data should be a list to allow expansion */
- pn_data_put_list (properties);
+ /* Container for application properties is a map */
+ pn_data_put_map(properties);
/* Enter into the list for access to its elements */
pn_data_enter (properties);
@@ -116,8 +116,8 @@ qpidBridgeMsgCodec_pack (msgBridge bridgeMessage,
{
return status;
}
- pn_data_put_ubyte (properties, type);
-
+ pn_data_put_string (properties,pn_bytes(strlen(QPID_KEY_MSGTYPE),QPID_KEY_MSGTYPE));
+ pn_data_put_ubyte (properties,type);
switch (type)
{
@@ -129,14 +129,16 @@ qpidBridgeMsgCodec_pack (msgBridge bridgeMessage,
{
return status;
}
+ pn_data_put_string (properties,pn_bytes(strlen(QPID_KEY_INBOXNAME),QPID_KEY_INBOXNAME));
pn_data_put_string (properties, pn_bytes (strlen(inboxName),
- inboxName));
+ inboxName));
status = qpidBridgeMamaMsgImpl_getReplyTo (bridgeMessage, &replyTo);
if (MAMA_STATUS_OK != status)
{
return status;
}
+ pn_data_put_string (properties,pn_bytes(strlen(QPID_KEY_REPLYTO),QPID_KEY_REPLYTO));
pn_data_put_string (properties, pn_bytes (strlen(replyTo),
replyTo));
@@ -149,6 +151,7 @@ qpidBridgeMsgCodec_pack (msgBridge bridgeMessage,
{
return status;
}
+ pn_data_put_string (properties,pn_bytes(strlen(QPID_KEY_TARGETSUBJECT),QPID_KEY_TARGETSUBJECT));
pn_data_put_string (properties, pn_bytes (strlen(targetSubject),
targetSubject));
break;
@@ -253,13 +256,23 @@ qpidBridgeMsgCodec_unpack (msgBridge bridgeMessage,
/* Skip over the initial null atom */
pn_data_next (properties);
- /* Main container should be a list to allow expansion */
- pn_data_get_list (properties);
+ /* Container for application properties is a map */
+ pn_data_get_map (properties);
pn_data_enter (properties);
- /* Get the message type out */
- pn_data_next (properties);
- type = (qpidMsgType) pn_data_get_ubyte (properties);
+ int found = 0;
+ found = pn_data_lookup (properties,QPID_KEY_MSGTYPE);
+ if (found)
+ {
+ type = (qpidMsgType) pn_data_get_ubyte (properties);
+ }
+ else
+ {
+ mama_log (MAMA_LOG_LEVEL_ERROR,
+ "qpidBridgeMamaMsgImpl_unpack(): "
+ "Unable to retrieve message type from message");
+ return MAMA_STATUS_PLATFORM;
+ }
qpidBridgeMamaMsgImpl_setMsgType (bridgeMessage, type);
@@ -268,16 +281,38 @@ qpidBridgeMsgCodec_unpack (msgBridge bridgeMessage,
case QPID_MSG_INBOX_REQUEST:
/* Move onto inbox name and extract / copy */
- pn_data_next (properties);
- prop = pn_data_get_string (properties);
+ found = pn_data_lookup (properties,QPID_KEY_INBOXNAME);
+ if (found)
+ {
+ prop = pn_data_get_string (properties);
+ }
+ else
+ {
+ mama_log (MAMA_LOG_LEVEL_ERROR,
+ "qpidBridgeMamaMsgImpl_unpack(): "
+ "Unable to retrieve inbox name");
+ return MAMA_STATUS_PLATFORM;
+ }
+
status = qpidBridgeMamaMsgImpl_setInboxName (bridgeMessage, prop.start);
if (MAMA_STATUS_OK != status)
{
return status;
}
- /* Move onto reply to url and extract / copy */
- pn_data_next (properties);
- prop = pn_data_get_string (properties);
+
+ found = pn_data_lookup (properties,QPID_KEY_REPLYTO);
+ if (found)
+ {
+ prop = pn_data_get_string (properties);
+ }
+ else
+ {
+ mama_log (MAMA_LOG_LEVEL_ERROR,
+ "qpidBridgeMamaMsgImpl_unpack(): "
+ "Unable to retrieve reply to url");
+ return MAMA_STATUS_PLATFORM;
+ }
+
status = qpidBridgeMamaMsgImpl_setReplyTo (bridgeMessage, prop.start);
if (MAMA_STATUS_OK != status)
{
@@ -285,9 +320,20 @@ qpidBridgeMsgCodec_unpack (msgBridge bridgeMessage,
}
break;
case QPID_MSG_INBOX_RESPONSE:
- /* Move onto target subject and extract / copy */
- pn_data_next (properties);
- prop = pn_data_get_string (properties);
+
+ found = pn_data_lookup (properties,QPID_KEY_TARGETSUBJECT);
+ if (found)
+ {
+ prop = pn_data_get_string (properties);
+ }
+ else
+ {
+ mama_log (MAMA_LOG_LEVEL_ERROR,
+ "qpidBridgeMamaMsgImpl_unpack(): "
+ "Unable to retrieve target container");
+ return MAMA_STATUS_PLATFORM;
+ }
+
status = qpidBridgeMamaMsgImpl_setTargetSubject (bridgeMessage,
prop.start);
if (MAMA_STATUS_OK != status)
diff --git a/mama/c_cpp/src/c/bridge/qpid/publisher.c b/mama/c_cpp/src/c/bridge/qpid/publisher.c
index 998b145..4d32475 100644
--- a/mama/c_cpp/src/c/bridge/qpid/publisher.c
+++ b/mama/c_cpp/src/c/bridge/qpid/publisher.c
@@ -587,12 +587,13 @@ qpidBridgePublisherImpl_setMessageType (pn_message_t* message, qpidMsgType type)
/* Ensure position is at the start */
pn_data_rewind (properties);
- /* Main container should be a list to allow expansion */
- pn_data_put_list (properties);
- pn_data_enter (properties);
+ /* Container for application properties is a map */
+ pn_data_put_map(properties);
+ pn_data_enter(properties);
/* Add the type */
- pn_data_put_ubyte (properties, type);
+ pn_data_put_string(properties,pn_bytes(strlen(QPID_KEY_MSGTYPE),QPID_KEY_MSGTYPE));
+ pn_data_put_ubyte(properties,type);
pn_data_exit (properties);
diff --git a/mama/c_cpp/src/c/bridge/qpid/qpiddefs.h b/mama/c_cpp/src/c/bridge/qpid/qpiddefs.h
index 8e686ea..6c1adf4 100644
--- a/mama/c_cpp/src/c/bridge/qpid/qpiddefs.h
+++ b/mama/c_cpp/src/c/bridge/qpid/qpiddefs.h
@@ -99,6 +99,11 @@ typedef enum qpidMsgType_
#endif /* _PROTON_VERSION_H */
+/* Keys for application property map */
+#define QPID_KEY_MSGTYPE "MAMAT"
+#define QPID_KEY_INBOXNAME "MAMAI"
+#define QPID_KEY_REPLYTO "MAMAR"
+#define QPID_KEY_TARGETSUBJECT "MAMAS"
/*=========================================================================
= Typedefs, structs, enums and globals =
diff --git a/mama/c_cpp/src/c/bridge/qpid/transport.c b/mama/c_cpp/src/c/bridge/qpid/transport.c
index 8ca9550..ba21810 100644
--- a/mama/c_cpp/src/c/bridge/qpid/transport.c
+++ b/mama/c_cpp/src/c/bridge/qpid/transport.c
@@ -1411,11 +1411,22 @@ void* qpidBridgeMamaTransportImpl_dispatchThread (void* closure)
/* Move to the first element inside */
pn_data_next (properties); /* Move past first NULL byte */
- pn_data_enter (properties); /* Enter into meta list */
- pn_data_next (properties); /* Next lines up first entry */
+ pn_data_get_map(properties);
+ pn_data_enter (properties); /* Enter into meta map */
- /* Pull out the packet type */
- msgNode->mMsgType = (qpidMsgType) pn_data_get_ubyte (properties);
+ int found = 0;
+ found = pn_data_lookup(properties,QPID_KEY_MSGTYPE);
+ if (found)
+ {
+ msgNode->mMsgType = (qpidMsgType) pn_data_get_ubyte (properties);
+ }
+ else
+ {
+ mama_log (MAMA_LOG_LEVEL_ERROR,
+ "qpidBridgeMamaTransportImpl_dispatchThread(): "
+ "Unable to retrieve message type from message");
+ return NULL;
+ }
switch (msgNode->mMsgType)
{
--
1.8.5.2 (Apple Git-48)
