Skip to main content

"getMessagesFromMqttSession"

Use the "getMessagesFromMqttSession" action to retrieve undelivered messages from a client-subscribed MQTT topic

The "getMessagesFromMqttSession" action retrieves undelivered messages from an MQTT topic to which an MQTT client has subscribed. 

The MQTT client is identified by the required "clientName" property. An MQTT session with that client identifier must already exist; otherwise, it returns an error.

Each time you call this action with the same "clientName", the server starts where it left off and delivers the next set of messages published to that client. It retrieves undelivered messages up to the number specified by "maxMessages". The message payload is encoded according to the  "binaryFormat" property.

If you subscribe to multiple topics, the server returns messages from all of them. You can use the "topic" property to retrieve messages from only one of your subscribed topics. 

The typical pattern for using this action is to first call the "createMqttSession" action to create a new session. Then use the "subscribeToMqtt" action to subscribe to topics. You can then repeatedly call the "getMessagesFromMqttSession" action to retrieve messages as clients publish them. 

An application can retrieve messages sent to a shared topic, which is a topic starting with $share/. A shared topic works differently from a normal topic because the server distributes each message to a different subscriber rather than sending each message to each subscriber. Each time a subscriber uses the "getMessagesFromMqttSession" action to retrieve a message, the server removes the message from the queue and prevents that message from being sent to another subscriber. If the application encounters an error processing a message, it must publish the message back to the topic so it can be processed by another subscriber. 

Tip

An application can use "getMessagesFromMqttTopic" to retrieve shared messages without removing them from the queue.

Warning

MQTT clients must have unique client names so they do not consume each other's messages. You must take extra care when using the stateful "getMessagesFromMqttSession" action to ensure that no other process is using the same MQTT client identifier; otherwise, each process that retrieves messages denies other processes from retrieving them. If you want to use the event streaming technique to retrieve messages, do not subscribe to topics; instead, use the "getMessagesFromMqttTopic" action to retrieve messages.

Optionally, set  "includeMqttProperties": true to include the message's MQTT properties. It defaults to false.

Optionally, set "includeTransformFields": true to include the custom fields used for the transformation process's inputs and outputs. It defaults to false.

Request examples

Maximal request

{
  "api": "mq",
  "action": "getMessagesFromMqttSession",
  "params":
  {
    "topic": "optional topic name",
    "clientName": "required unique MQTT client identifier",
    "maxMessages": 100,
    "binaryFormat": "base64 | hex | utf8 | json | byteArray",
    "includeMessageMetadata": true,
    "includeMqttProperties": true,
    "includeTransformFields": true
  },
  "apiVersion": "1.0",
  "debug": "max",
  "requestId": "optionalUniqueRequestIdFromTheClient",
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve the next batch of 100 messages

{
  "api": "mq",
  "action": "getMessagesFromMqttSession",
  "params":
  {
    "clientName": "required unique MQTT client identifier",
    "maxMessages": 100
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve all undelivered messages

{
  "api": "mq",
  "action": "getMessagesFromMqttSession",
  "params":
  {
    "clientName": "required unique MQTT client identifier"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve up to 100 unsent messages from the subscription, format the message payload as JSON, and include all expired messages, MQTT metadata, MQT properties, and transform fields

{
  "api": "mq",
  "action": "getMessagesFromMqttSession",
  "params":
  {
    "clientName": "required unique MQTT client identifier",
    "topicFilter": "topic#",
    "maxMessages": 100,
    "binaryFormat": "json",

    "includeMessageMetadata": true,
    "includeMqttProperties": true,
    "includeTransformFields": true
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Response example

{
  "result": {

    "messages":
    [
      {
        "clientName": "MQTT client identifier in mqtt_client_id",
        "topic": "topic1",
        "subscriptionIdentifiers": [123],
        "databaseName": "faircom",
        "ownerName": "admin",
        "tableName": "mqtt_msg_topic1",

        "payload": "C3A3",
        "binaryFormat": "base64 | hex | utf8 | json | byteArray",

        "id": "263",
        "timestamp": "2023-12-18T18:04:20:085",
        "error": false,
        "log": null,

        "contentType": "IanaMediaType or a CustomType",
        "correlationData": "any JSON value up to 65500 bytes",
        "expirySeconds": 60,
        "messageHasExpired": true,
        "payloadFormat": "binary | utf8",
        "publishedQoS": 2,
        "retain": true,
        "responseTopic": "topic/name/for/subscriber/to/respond/to",
        "userProperties": [
          {
            "key": "some key",
            "value": "some value"
          }
        ], 

        "faircomClientId": 6732,
        "faircomTopicId": 457,
        "faircomForwardTopicId": 0,
        "faircomMqttPacketId": 53736,
        "faircomMqttSentReason": "Client | Retained | Will | Forwarded",

        "transformFields": [
          { "fieldName": "t1", "fieldData": "some value" }
        ]

      }
    ]

  },
  "errorCode": 0,
  "errorMessage": "",
  "debugInfo": {
  },
  "requestId": "optionalUniqueRequestIdFromTheClient",
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Use the "getMessagesFromMqttSession" action to retrieve undelivered messages from a client-subscribed MQTT topic

getMessagesFromMqttSessionget messages from mqtt sessionjsonActionJSON MQ message API

Params

Table 1. Params property summaries

Property

Description

Default

Type

Limits (inclusive)

binaryFormat

specifies how the "payload" property is encoded

"json"

enum

"base64"
"hex"
"byteArray"
"utf8"
"json"

clientName

specifies the unique name that identifies the client to the FairCom MQTT engine

""

string

0 to 65,550 bytes

includeMessageMetadata

returns the message's "id", "timestamp", "error", and "log" properties in the response

false

Boolean

true
false

includeMqttProperties

returns MQTT properties in the response

false

Boolean

true
false

includeTransformFields

returns custom transform fields in the response

false

Boolean

true
false

maxMessages

specifies the number of messages to return from a "getMessages..." action

20

integer

0 to 100000

topic

specifies an MQTT topic

""

UTF-8 string



"binaryFormat"

The "binaryFormat" property specifies how the "payload" property is encoded. It also applies to the format of all binary fields of messages included in a response. 

It defaults to "json". When the MQTT payload is not JSON, the API returns the following error message: "MQTT payload cannot be encoded as JSON. Set binaryFormat to base64 or hex to retrieve the payload's binary value." 

Note

In other FairCom APIs, the default value of "binaryFormat" is "base64". In this API, it defaults to "json" because JSON is the most common payload in MQTT messages.

You may set "binaryFormat" to one of the following values:

  • "base64" tells the server you set the payload to a string containing a base64 encoded value.

  • "hex" tells the server you set the payload to a string containing a hexadecimal encoded value.

  • "byteArray" tells the server you set the payload to an array of integers. Each integer must be between 0 and 255, representing one byte in the payload.

  • "utf8" tells the server you set the payload to a string. The server reads the UTF-8 encoded bytes in the string and directly stores them in the payload without conversion. 

    • The value stored in the payload does not include the double quotes before and after the string. 

    • For example, "payload": "I am a string", is stored in the payload as I am a string. Notice how the server does not store the double quotes.

  • "json" tells the server you set the payload to a JSON value. A JSON value may be one of the following:

    • A JSON object assigned to the payload, such as "payload": { "am": "an object" }, causes the server to store the UTF-8 characters { "am": "an object" } in the payload. 

      • The server must store the exact highlighted text, including the curly braces. 

      • The server may alter the object's whitespace, such as {"am":"an object"}.

    • A JSON array assigned to the payload, such as "payload": [ "am", "array" ], causes the server to store the UTF-8 characters [ "am", "array" ] in the payload. 

      • The server must store the exact highlighted text, including the straight braces.

      • The server may alter the object's whitespace, such as ["am","array"].

    • A JSON string assigned to the payload, such as "payload": "I am a JSON string", causes the server to store the UTF-8 characters "I am a JSON string" in the payload. 

      • The server must store the exact highlighted text, including the double quotes because a JSON string type includes the double-quote characters before and after the string's value.

    • A JSON number assigned to the payload, such as "payload": -12.34, causes the server to store the UTF-8 characters -12.34 in the payload.

    • A JSON true assigned to the payload, such as "payload": true, causes the server to store the UTF-8 characters true in the payload.

    • A JSON false assigned to the payload, such as "payload": false, causes the server to store the UTF-8 characters false in the payload.

    • A JSON null assigned to the payload, such as "payload": null, causes the server to store the UTF-8 characters null in the payload.

"clientName"

The "clientName" property is an optional string of up to 65550 characters containing a client name that uniquely identifies the client to the FairCom MQTT engine. 

The MQTT protocol allows only one connection per client identifier because each connection is stateful. If another client with the same "clientName" is already connected, the MQTT broker disconnects the other client and vice-versa. 

The JSON MQ Message API behaves differently because it is stateless; it allows multiple processes to simultaneously use the same client identifier to manage sessions, subscribe to topics, send messages to topics, and retrieve messages from topics.

"includeMessageMetadata"

Set the "includeMessageMetadata" property to true to include the message's "id", "timestamp", "error", and "log" properties.

"includeMqttProperties"

Set the "includeMqttProperties" property to true to include the MQTT properties in the response. Otherwise, omit the property or set it to false or null. MQTT properties are helpful when you need to troubleshoot MQTT messages.

"includeTransformFields"

Set the "includeTransformFields" property to true to include custom transform fields (if any) that a customer added to the integration table. Otherwise, omit the property or set it to false or null. Custom transform fields are helpful when you want to see the outputs and inputs of transformations assigned to the integration table.maxMessages.

The "maxMessages" property is optional and specifies the maximum number of messages the server returns. When omitted or set to "null", it defaults to 20.

"maxMessages"

The "maxMessages" property is optional and defaults to 20. It has a minimum value of 1 and a maximum value of 100000. It specifies the number of messages to return from a "getMessages…" action. The number of returned messages will be equal or less than this maximum.

"topic"

The "topic" property is an optional UTF-8 string containing an MQTT topic. It may not MQTT wildcard characters # and +. It may contain multiple / characters, which segment an MQTT topic into a set of positional properties.