AMQP Messages
AMQP Types
AMQP defines a set of commonly used primitive types aimed towards a cross-platform interoperable data representation. These primitive types are divided into the following categories:
- Primitive
- Described
- Composite
- Restricted
NOTE: In Universal Messaging, an embedded Apache Qpid Proton-J protocol engine is used for type marshalling so all type mappings occur as defined by the Proton-J API. See the Qpid Proton page at https://qpid.apache.org/proton/index.html for details.
AMQP Message Format
AMQP offers a standard message format (default) which can be overridden by an AMQP node. It consists of a bare message (immutable end-end) and an annotated message which may be altered by peer / intermediate nodes.
The following figure illustrates the default AMQP message format:
NOTE: Universal Messaging supports the default message format and provides a transformation layer based on it. Currently no tests have been performed with a custom AMQP message format, at the very least transformers would fail.
Transfer Headers
| AMQP header | UM Header | AMQP Type | Java Type | UM |
|---|---|---|---|---|
| durable | nPublished.isPersistent | boolean | boolean |
|
| priority | nHeader.Priority | ubyte (Default: 4) | byte |
|
| ttl | nHeader.TTL | milliseconds (uint) | long |
|
| first-acquirer | nEventProperties.JMS_AMQP_FirstAcquirer | boolean | boolean |
|
| delivery-count | nHeader.RedeliveredCount | uint (Default: 0) | long |
|
Delivery Annotations
AMQP Delivery annotations offer an optional non-normative section for AMQP vendor defined recipients by typically providing delivery related information from the sending node to the receiving node.
| AMQP Delivery Annotation | UM Header Key | AMQP Key/Value Type | Java Key/Value Type | UM |
|---|---|---|---|---|
| SomeDeliveryAnnotation | JMS_AMQP_DA_<AMQP_SYMBOL_NAME(SomeDeliveryAnnotation)> | AMQP Symbol / AMQP Value | String / nEventProperties Value |
|
Universal Messaging currently does not make use of any delivery annotations but will preserve any received. The naming convention used is <JMS_AMQP_DA_><delivery annotation symbol>
Message Annotations
AMQP Message annotations offer an optional section for AMQP properties aimed at AMQP infrastructure nodes. The following table illustrates the mappings to Universal Messaging:
| AMQP Message Annotation | UM Header Key | AMQP Key/Value Type | Java Key/Value Type | UM |
|---|---|---|---|---|
| SomeMessageAnnotation | JMS_AMQP_MA_<AMQP SYMBOL_NAME(SomeMessageAnnotation)> | AMQP Symbol / AMQP Value | String / nEventProperties Value | |
| x-opt-jms-msg-type | JMS_AMQP_MA_x-opt-jms-msg-type | x-opt-jms-msg-type / AMQP byte | String / byte (nHeader.MessageType) |
|
| x-opt-jms-dest | JMS_AMQP_MA_x-opt-jms-dest | x-opt-jms-dest / AMQP String | String / byte[] (nHeader.Type) |
|
| x-opt-reply-type | JMS_AMQP_MA_x-opt-reply-type | x-opt-reply-type / AMQP byte | String / byte (nHeader.ReplyType) |
|
Standard Message Properties
AMQP standard message properties is a section used for a defined set of immutable standard properties of the message. The section is part of the bare message; therefore, if retransmitted by an intermediary, it must remain unaltered.
The following table illustrates how these standard message properties are mapped to Universal Messaging:
| AMQP Standard Property | UM Header Mapping | AMQP Type | Java Value Type | UM |
|---|---|---|---|---|
| message-id | nHeader.MessageId | AMQP String | byte[] |
|
| user-id | nHeader.UserId | AMQP Binary | byte[] |
|
| to | nHeader.Destination | AMQP String | byte[] |
|
| subject | nEventProperties.JMS_AMQP_SUBJECT | AMQP String | String |
|
| reply-to | nHeader.ReplyToName | AMQP String | byte[] |
|
| correlation-id | nHeader.CorrelationId | AMQP Value | byte[] |
|
| content-type | nEventProperties.JMS_AMQP_CONTENT_TYPE | AMQP Symbol | String |
|
| content-encoding | nEventProperties.JMS_AMQP_CONTENT_ENCODING | AMQP Symbol | String |
|
| absolute-expiry-time | N/A (Currently unsupported) | AMQP Date | long |
|
| creation-time | nHeader.Timestamp | AMQP Date | long |
|
| group-id | nEventProperties.JMSXGroupID | AMQP String | String |
|
| group-sequence | nEventProperties.JMS_AMQP_GROUP_SEQUENCE | AMQP uint | int |
|
| reply-to-group-id | nEventPropties.JMS_AMQP_ReplyToGroupID | AMQP String | String |
|
Application Message Properties
AMQP offers an optional application-properties section, part of the bare message used for structured application data. Intermediaries can use the data within this structure for the purposes of filtering or routing.
The keys of this map are restricted to be of type string (which
excludes the possibility of a null key) and the values are restricted to be of
simple types only, that is, excluding map,
list, and array types. The following table
illustrates how these are mapped to Universal Messaging:
| AMQP Application Message Properties | UM Dictionary Key | AMQP Key/Value Type | Java Key/Value Type | UM |
|---|---|---|---|---|
| MAP < STRING, SIMPLE VALUE > | AMQP String / AMQP Value | String / nEventProperties Value |
|
|
| example_application_key1 | example_application_key1 | AMQP String / AMQP Value | String / nEventProperties Value |
|
Message Body
The AMQP message body is defined in three ways:
- One or more AMQP Data sections
- One or more AMQP Sequence sections
- A single AMQP Value section
Universal Messaging currently aims to offer support for JMS 1.1 use cases via AMQP and therefore has implemented some mappings via the Complete AMQP transformer.
The mappings are based on the following UM constants:
| UM nHeader.MessageType | UM Dictionary Key (transformed) | Value (byte) | UM |
|---|---|---|---|
| JMS_BASE_MESSAGE_TYPE | AMQP_Type | 0 |
|
| JMS_MAP_MESSAGE_TYPE | AMQP_Type | 1 |
|
| JMS_BYTES_MESSAGE_TYPE | AMQP_Type | 2 |
|
| JMS_OBJECT_MESSAGE_TYPE | AMQP_Type | 3 |
|
| JMS_STREAM_MESSAGE_TYPE | AMQP_Type | 4 |
|
| JMS_TEXT_MESSAGE_TYPE | AMQP_Type | 5 |
|
Universal Messaging extended message types, such as protobuf, are not supported by the transformation framework.
Additionally, mappings (depending on the direction) are based on the following UM AMQP Transformation constants:
| UM AMQP transformation Constant | UM Dictionary Key (transformed) | Value (int) | UM |
|---|---|---|---|
| sAMQPData | AMQP_Type | 0 |
|
| sAMQPValue | AMQP_Type | 1 |
|
| sAMQPList | AMQP_Type | 2 |
|
| sAMQPEmpty | AMQP_Type | 3 |
|
| sAMQPUnknown | AMQP_Type | -1 |
|
Message Body: AMQP payloads to UM Native payloads
Initially, the Universal Messaging realm will check for an AMQP message annotation x-opt-jms-msg-type on the AMQP message. If detected and within the range of the above mentioned table, then the following conversions are performed:
| x-opt-jms-msg-type Message Annotation | AMQP Body | UM Stamping | UM payload | UM |
|---|---|---|---|---|
| JMS_BASE_MESSAGE_TYPE | Any | AMQP_Type=sAMQPEmpty | byte[0] |
|
| JMS_BYTES_MESSAGE_TYPE | null | byte[0] |
|
|
| JMS_BYTES_MESSAGE_TYPE | Data | AMQP_Type=sAMQPData | Binary.Array |
|
| JMS_BYTES_MESSAGE_TYPE | AMQP Value <Binary> | AMQP_Type=sAMQPValue | Binary.Array |
|
| JMS_TEXT_MESSAGE_TYPE | null | byte[0] |
|
|
| JMS_TEXT_MESSAGE_TYPE | Data | AMQP_Type=sAMQPData | Binary.Array (UTF-8) |
|
| JMS_TEXT_MESSAGE_TYPE | AMQP Value <Binary> | AMQP_Type=sAMQPValue | Binary.Array (UTF-8) |
|
| JMS_OBJECT_MESSAGE_TYPE | null | AMQP_Type=sAMQPData | byte[0] |
|
| JMS_OBJECT_MESSAGE_TYPE | Data | AMQP_Type=sAMQPData | Binary.Array |
|
| JMS_STREAM_MESSAGE_TYPE | null | byte[0] |
|
|
| JMS_STREAM_MESSAGE_TYPE | AMQP Value <List> | AMQP_Type=sAMQPValue | byte[](Serialized Vector<Binary>) |
|
| JMS_STREAM_MESSAGE_TYPE | AMQP Sequence | AMQP_Type=sAMQPList | byte[](Serialized Vector<Binary>) |
|
| JMS_MAP_MESSAGE_TYPE | null | byte[0] |
|
|
| JMS_MAP_MESSAGE_TYPE | AMQP Value <Map> | AMQP_Type=sAMQPValue | byte[](Externalized fEventDictionary) |
|
If the message annotation is not present, then the mapping starts looking for payload hints in message properties as follows:
| AMQP Message Property | AMQP Body | UM nHeader.MessageType | UM payload | UM |
|---|---|---|---|---|
| content-type | ||||
| text/plain | null | JMS_TEXT_MESSAGE_TYPE | byte[0] |
|
| application/x-java-serialized-object | null | JMS_OBJECT_MESSAGE_TYPE | byte[0] |
|
| application/octet-stream | null | JMS_BYTES_MESSAGE_TYPE | byte[0] |
|
| null or other | null | JMS_BASE_MESSAGE_TYPE | byte[0] |
|
| text/plain | Data | JMS_TEXT_MESSAGE_TYPE | Binary.Array(UTF-8) |
|
| application/x-java-serialized-object | Data | JMS_OBJECT_MESSAGE_TYPE | Binary.Array |
|
| application/octet-stream or null | Data | JMS_BYTES_MESSAGE_TYPE | Binary.Array |
|
| other | Data | JMS_BYTES_MESSAGE_TYPE | Binary.Array |
|
| AMQP Value <Type> body | ||||
| null | AMQP Value <null> | JMS_BASE_MESSAGE_TYPE | byte[0] |
|
| String | AMQP Value <String> | JMS_TEXT_MESSAGE_TYPE | Binary.Array (UTF-8) |
|
| Binary | AMQP Value <Binary> | JMS_BYTES_MESSAGE_TYPE | Binary.Array |
|
| List | AMQP Value <List> | JMS_STREAM_MESSAGE_TYPE | byte[](Serialized Vector<Binary>) |
|
| Map | AMQP Value <Map> | JMS_MAP_MESSAGE_TYPE | byte[](Externalized fEventDictionary) |
|
| other | AMQP Value <Object> | JMS_OBJECT_MESSAGE_TYPE | Binary.Array |
|
| AMQP Sequence <Object> body | ||||
| Binary | AMQP Sequence<Binary> | JMS_OBJECT_MESSAGE_TYPE | Binary.Array |
|
Message Body: UM Native Payloads to AMQP payloads
Initially, the Universal Messaging realm will check for an nEventProperties keys called AMQP_Type which is expected to be within the range defined by the AMQP Transformation Constants table. Typically these are AMQP messages previously transformed to native but as there are multiple possibilities with regard to the type of payload this should be transformed to, this value gives us a hint. If not present, it defaults to sAMQPUnknown.
Based on the UM nHeader.MessageType values the following conversions are performed:
| UM nHeader.MessageType | UM Dictionary Key AMQP_Type | AMQP Payload | UM |
|---|---|---|---|
| JMS_BASE_MESSAGE_TYPE | sAMQPEmpty or sAMQPUnknown or sAMQPData | Data(byte[0]) |
|
| JMS_BASE_MESSAGE_TYPE | sAMQPValue | AMQPValue(null) |
|
| JMS_BASE_MESSAGE_TYPE | sAMQPList |
|
|
| JMS_MAP_MESSAGE_TYPE | sAMQPEmpty | Data(byte[0]) |
|
| JMS_MAP_MESSAGE_TYPE | sAMQPValue or sAMQPUnknown | AMQPValue<Map> (from byte[](Externalized fEventDictionary) |
|
| JMS_MAP_MESSAGE_TYPE | sAMQPList or sAMQPData |
|
|
| JMS_BYTES_MESSAGE_TYPE | sAMQPEmpty | Data(byte[0]) |
|
| JMS_BYTES_MESSAGE_TYPE | sAMQPData orsAMQPUnknown | Data(Binary(byte[])) (from UM native payload) |
|
| JMS_BYTES_MESSAGE_TYPE | sAMQPValue | AMQPValue(Binary(byte[])) (from UM native payload) |
|
| JMS_BYTES_MESSAGE_TYPE | sAMQPList |
|
|
| JMS_OBJECT_MESSAGE_TYPE | sAMQPEmpty | Data(byte[0]) |
|
| JMS_OBJECT_MESSAGE_TYPE | sAMQPData orsAMQPUnknown | Data(Binary(byte[])) (from UM native payload) |
|
| JMS_OBJECT_MESSAGE_TYPE | sAMQPList or sAMQPValue |
|
|
| JMS_STREAM_MESSAGE_TYPE | sAMQPEmpty | Data(byte[0]) |
|
| JMS_STREAM_MESSAGE_TYPE | sAMQPList | AMQPSequence<Vector> (byte[](Serialized Vector<Binary>) |
|
| JMS_STREAM_MESSAGE_TYPE | sAMQPValue or sAMQPUnknown | AMQPValue<Vector> (byte[](Serialized Vector<Binary>) |
|
| JMS_STREAM_MESSAGE_TYPE | sAMQPData |
|
|
| JMS_TEXT_MESSAGE_TYPE | sAMQPEmpty | Data(byte[0]) |
|
| JMS_TEXT_MESSAGE_TYPE | sAMQPData | Data(Binary(byte[])) (from UM native payload) |
|
| JMS_TEXT_MESSAGE_TYPE | sAMQPValue orsAMQPUnknown | AMQPValue<String> (from UM native payload) |
|
| JMS_TEXT_MESSAGE_TYPE | sAMQPList |
|
Message Footer
AMQP Message footer offers an optional non-normative section for delivery or message metadata that can be calculated / validated only after the whole message has been processed (e.g digital signature verification, encryption etc).
The following table shows how these are mapped in Universal Messaging:
| AMQP Message Footer | UM Sub Dictionary Key | AMQP Key/Value Type | Java Key/Value Type | UM |
|---|---|---|---|---|
| MAP < SYMBOL, VALUE > | Footer | AMQP Symbol / AMQP Value | String / nEventProperties Value |
|
| example_footer_key1 | Footer.example_footer_key1 | AMQP Symbol / AMQP Value | String / nEventProperties Value |
|