`"getMessagesFromMqTopic"` (MQ API)

Use the `"getMessagesFromMqTopic"` action to retrieve messages published on a specific topic

This action uses the event streaming technique to retrieve messages published on a specific topic by the MQ API and MQTT. It is stateless and does not remove messages from the topic's queue. Use it to tail the end of the queue, retrieve all messages from the beginning of the queue, retrieve messages starting from a timestamp, or paginate through a queue.

This action allows multiple processes to retrieve messages simultaneously from a topic without moving the client's message position or removing shared messages.

Instead of using a client's message position to retrieve undelivered messages, it retrieves messages from a specified starting location in the topic's message queue. You can retrieve messages from the end or beginning of the queue, a specific timestamp, or a queue position ID.

This action does not disconnect MQTT clients.

Call this action to retrieve the "maxMessages" number of messages from a topic starting from the position specified by "startFrom". The message payload is encoded according to "binaryFormat".

You must use one but not both of the following properties to identify the message queue from which you want to retrieve messages:

Use "topic" to specify a topic from which you want to retrieve messages. A topic cannot contain wildcards.
Use "tableName" to specify the name of an integration table containing messages. You may also set the table's "databaseName" and "ownerName"; otherwise, the server will use the JSON Action Session's default values.

The optional "startFrom" property controls the starting point where the server delivers messages.

Set "startFrom" to the default of ["tail"] to retrieve the most recently sent messages.
Set "startFrom" to "head" to retrieve messages from the beginning.
Set "startFrom" to an ISO 8601 DateTime, such as "2024-07-11T21:18:01.000", to retrieve messages starting on or after that date and time. If the time component is missing, it defaults to "00:00:00.000".
Set "startFrom" to an integer number, such as 52233, to retrieve messages starting with the message with that ID number.

Tip

To retrieve the next set of messages, set "startFrom" to one plus the ID number of the last message you received.

Optionally, set "expiredMessagesFilter" to "include", "exclude", or "only" to include or exclude expired messages from the results. Use "only" to include only expired messages in the results. It defaults to "include".

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.

Tip

For more query options, use the JSON DB "getRecords..." actions. You can use the "describeMqSession" to return information about a client's subscriptions, which includes a map of topic names to table names. You can query those tables to see their messages. You can use SQL to run ad hoc queries. You can find a record using partial, exact, or inequality matches. You can move and skip forward and backward through the records in index-sorted order.

Request examples

Minimal

{
  "api": "mq",
  "action": "getMessagesFromMqTopic",
  "params": {
    "topic": "test/topic1"
  }
}

Maximal

{
  "api": "mq",
  "action": "getMessagesFromMqTopic",
  "params":
  {
    "topic": "test/topic1",

    "databaseName": "faircom",
    "ownerName": "admin",
    "tableName": "mqtt_msg_test_topic1",

    "startFrom": number | "tail | head",
    "maxMessages": 1,
    "binaryFormat": "base64 | hex | utf8 | json | byteArray",

    "expiredMessagesFilter": "include",

    "includeMessageMetadata": true,
    "includeMqttProperties": true,
    "includeTransformFields": true,
    "includeOtherFields": true
  },
  "apiVersion": "1.0",
  "debug": "max",
  "requestId": "optionalUniqueRequestIdFromTheClient",
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve the most recently sent message to topic1 with all metadata and the message payload formatted as JSON

{
  "api": "mq",
  "action": "getMessagesFromMqTopic",
  "params":
  {
    "maxMessages": 1,
    "binaryFormat": "json",

    "topic": "test/topic1",
    "expiredMessagesFilter": "include",

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

Retrieve the last 10 messages sent to topic1 with the message payload formatted as base64 and no metadata

{
  "api": "mq",
  "action": "getMessagesFromMqTopic",
  "params":
  {
    "topic": "test/topic1",

    "startFrom": "tail",
    "maxMessages": 10,
    "binaryFormat": "base64"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve the last message sent to all topics, format the message payload as hexadecimal, and include all MQTT properties

{
 "api": "mq",
 "action": "getMessagesFromMqTopic",
 "params":
 {
   "topic": "#",
   "maxMessages": 1,
   "binaryFormat": "hex",

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

Retrieve the first 50 non-purged messages sent to topic1 with the message payload formatted as JSON

{
  "api": "mq",
  "action": "getMessagesFromMqTopic",
  "params":
  {
    "topic": "test/topic1",
    "startFrom": "head",
    "maxMessages": 50,
    "binaryFormat": "json"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve the next 50 non-purged messages sent to topic1

{
  "api": "mq",
  "action": "getMessagesFromMqTopic",
  "params":
  {
    "topic": "test/topic1",
    "startFrom": 51,
    "maxMessages": 50,
    "binaryFormat": "json"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve 5000 messages sent to topic1 on or after `"2024-07-11T21:18:01.000"`

{
  "api": "mq",
  "action": "getMessagesFromMqTopic",
  "params":
  {
    "topic": "test/topic1",
    "startFrom": "2024-07-11T21:18:01.000",
    "maxMessages": 5000,
    "binaryFormat": "json"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Response example

{
  "result": {

    "messages":
    [
      {
        "clientName": "Client identifier",
        "topic": "test/topic1",
        "subscriptionIdentifiers": [123],
        "databaseName": "faircom",
        "ownerName": "admin",
        "tableName": "mqtt_msg_test_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 `"getMessagesFromMqTopic"` action to retrieve messages published on a specific topic

getMessagesFromMqTopicget messages from mq topicjsonActionMQ API

`"params"`

Table 1. Params property summaries

Property

Description

Default

Type

Limits (inclusive)

binaryFormat

(optional) specifies how the "payload" property is encoded

"json"

string

One of the following: "base64", "hex", or "byteArray".

expiredMessagesFilter

(optional) specifies whether or not to include expired messages in the result

"include"

string

"include"

"exclude"

"only"

includeMessageMetadata

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

false

Boolean

true

false

includeMqttProperties

(optional) returns MQTT properties in the response

false

Boolean

true

false

includeOtherFields

(optional) returns additional fields that are not the built-in integration table, transform, and MQTT fields in the response

false

Boolean

true

false

includeTransformFields

(optional) returns custom transform fields in the response

false

Boolean

true

false

maxMessages

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

20

integer

0 to 100000

startFrom

(optional) sets the starting message position for a message filter

"tail"

string

"head"

"tail"

string ISO 8601 DateTime value

message ID number

topic

(optional) specifies the name of a topic

""

UTF-8 string

`"binaryFormat"`

The "binaryFormat" property designates how binary values are formatted in the JSON request and JSON response message. The default value is "base64".

Note

Typically, response options apply only to the server’s response, but the "binaryFormat" property applies to both the request and the response.

The "binaryFormat" property may occur inside "params", "responseOptions", "defaultResponseOptions" , and "result".
When "binaryFormat" occurs in "params" it specifies how the sender represents binary values.
For example, when "binaryFormat" is set to "hex", the FairCom server expects the binary values of fields and keys to be represented in strings with hexadecimal format.
When "binaryFormat" occurs in "responseOptions" or "defaultResponseOptions" it specifies how the FairCom server should represent binary values in responses.
For example, when "binaryFormat" is set to "hex", the FairCom server represents binary values in strings with hexadecimal format.
When "binaryFormat" occurs in "result", it signifies how binary values are represented.
For example, when "binaryFormat" is set to "base64", the FairCom server represents binary values in the response in base64 format.
The following are the possible values for each format.
- "base64"
  When the server reads and writes from a binary field, it represents the binary value as a base64 string.
  - Base64 is harder for people to read and convert to binary.
  - Base64 creates the smallest payload for the most efficient data transmission in JSON.
  - "base64" strings contain the characters 0 - 9 , A - Z, a - z, +, /, and =.
- "hex"
  When the server reads and writes from a binary field, it represents the binary value as a hexadecimal string.
  - Hexadecimal is easier for people to read and convert to binary.
  - Hexadecimal creates a 30% larger payload than "base64", which makes it less efficient for data transmission.
  - Hexadecimal strings contain the characters 0 - 9 and A - F.
- "byteArray"
  When the server reads and writes from a binary field, it represents the binary value as an array of bytes.
  - An array of bytes is easiest for a program to manipulate.
  - An array of bytes creates a larger payload than "base64" and "hex", which makes it less efficient for data transmission.
  - An array of bytes returns a JSON array containing one integer number between 0 and 255 for each byte in the binary value:
    "aBinaryField": [ 255, 0, 255 ]

Create a "binary_test" table

This example creates a table containing one binary field named "bin" with a fixed length of 5 bytes.

{
  "api": "db",
  "action": "createTable",
  "params": {
    "tableName": "binary_test",
    "fields": [
      {
        "name": "bin",
        "type": "binary",
        "length": 5
      }
    ]
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Insert a record into the "binary_test" table using an array of bytes format

This example inserts a record with the ASCII characters "123" in the "bin" field. The value of "bin" is represented as an array of bytes.

{
  "api": "db",
  "action": "insertRecords",
  "params": {
    "tableName": "binary_test",
    "dataFormat": "objects",
    "binaryFormat": "byteArray",
    "sourceData": [
      {
        "bin": [49,50,51]
      }
    ]
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Insert a record into the "binary_test" table using hexadecimal format

This example inserts a record with the ASCII characters "123" in the "bin" field. The value of "bin" is represented as a string in hexadecimal format.

{
  "api": "db",
  "action": "insertRecords",
  "params": {
    "tableName": "binary_test",
    "dataFormat": "objects",
    "binaryFormat": "hex",
    "sourceData": [
      {
        "bin": "313233"
      }
    ]
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Insert a record into the "binary_test" table using base64 format

This example inserts a record with the ASCII characters "123" in the "bin" field. The value of "bin" is represented as a string in base64 format.

{
  "api": "db",
  "action": "insertRecords",
  "params": {
    "tableName": "binary_test",
    "dataFormat": "objects",
    "binaryFormat": "base64",
    "sourceData": [
      {
        "bin": "MTIz"
      }
    ]
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve a record with "binaryFormat" as an array of bytes

This example requests the first record in the "binary_test" table with the value of "bin" represented as an array of bytes.

{
  "api": "db",
  "action": "getRecordsByTable",
  "params": {
    "tableName": "binary_test",
    "maxRecords": 1
  },
  "responseOptions": {
    "binaryFormat": "byteArray",
    "dataFormat": "objects",
    "numberFormat": "number"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Note

Our examples insert only 3 bytes into "bin". Because the "bin" field has a fixed length of 5 bytes, the server pads unused bytes with 0x00 and stores the result. When a record is retrieved, the server returns all 5 bytes.

{
  "result": {
    "dataFormat": "objects",
    "binaryFormat": "byteArray",
    "fields": [
      { "name": "id",       "type": "bigint", "length": null, "scale": null, "autoTimestamp": "none", "defaultValue": null, "nullable": false, "primaryKey": 1 },
      { "name": "changeId", "type": "bigint", "length": null, "scale": null, "autoTimestamp": "none", "defaultValue": null, "nullable": true,  "primaryKey": 0 },
      { "name": "bin",      "type": "binary", "length": 5,    "scale": null, "autoTimestamp": "none", "defaultValue": null, "nullable": true,  "primaryKey": 0 }
    ],
    "data": [
      {
        "bin": [49,50,51,0,0],
        "changeId": 50217,
        "id": 1
      }
    ],
    "moreRecords": true,
    "requestedRecordCount": 1,
    "returnedRecordCount": 1,
    "totalRecordCount": 3
  },
  "errorCode": 0,
  "errorMessage": "",
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve a record with "binaryFormat" as hexadecimal

This example requests the first record in the "binary_test" table with the value of "bin" represented as a hexadecimal string.

{
  "api": "db",
  "action": "getRecordsByTable",
  "params": {
    "tableName": "binary_test",
    "maxRecords": 1
  },
  "responseOptions": {
    "binaryFormat": "hex",
    "dataFormat": "objects",
    "numberFormat": "number"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

{
  "result": {
    "dataFormat": "objects",
    "binaryFormat": "byteArray",
    "fields": [
      { "name": "id",       "type": "bigint", "length": null, "scale": null, "autoTimestamp": "none", "defaultValue": null, "nullable": false, "primaryKey": 1 },
      { "name": "changeId", "type": "bigint", "length": null, "scale": null, "autoTimestamp": "none", "defaultValue": null, "nullable": true,  "primaryKey": 0 },
      { "name": "bin",      "type": "binary", "length": 5,    "scale": null, "autoTimestamp": "none", "defaultValue": null, "nullable": true,  "primaryKey": 0 }
    ],
    "data": [
      {
        "bin": "3132330000",
        "changeId": 50217,
        "id": 1
      }
    ],
    "moreRecords": true,
    "requestedRecordCount": 1,
    "returnedRecordCount": 1,
    "totalRecordCount": 3
  },
  "errorCode": 0,
  "errorMessage": "",
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

Retrieve a record with "binaryFormat" as base64

This example requests the first record in the "binary_test" table with the value of "bin" represented as a base64 string.

{
  "api": "db",
  "action": "getRecordsByTable",
  "params": {
    "tableName": "binary_test",
    "maxRecords": 1
  },
  "responseOptions": {
    "binaryFormat": "base64",
    "dataFormat": "objects",
    "numberFormat": "number"
  },
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

{
  "result": {
    "dataFormat": "objects",
    "binaryFormat": "byteArray",
    "fields": [
      { "name": "id",       "type": "bigint", "length": null, "scale": null, "autoTimestamp": "none", "defaultValue": null, "nullable": false, "primaryKey": 1 },
      { "name": "changeId", "type": "bigint", "length": null, "scale": null, "autoTimestamp": "none", "defaultValue": null, "nullable": true,  "primaryKey": 0 },
      { "name": "bin",      "type": "binary", "length": 5,    "scale": null, "autoTimestamp": "none", "defaultValue": null, "nullable": true,  "primaryKey": 0 }
    ],
    "data": [
      {
        "bin": "MTIzAAA=",
        "changeId": 50217,
        "id": 1
      }
    ],
    "moreRecords": true,
    "requestedRecordCount": 1,
    "returnedRecordCount": 1,
    "totalRecordCount": 3
  },
  "errorCode": 0,
  "errorMessage": "",
  "authToken": "replaceWithAuthTokenFromCreateSession"
}

`"expiredMessagesFilter"`

The "expiredMessagesFilter" property is optional. It is an enumerated string that contains one of the following values: "include", "exclude", and "only". It defaults to "include".

"include" (default) returns expired messages in the results.
"exclude" omits expired messages from the results.
"only" returns only expired messages in the results.

`"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.

`"includeOtherFields"`

Set the "includeOtherFields" property to true to include additional fields that are not the built-in integration table fields, transform fields, and MQTT fields. Otherwise, omit the property or set it to false or null.

`"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.

`"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 to or less than this maximum.

`"startFrom"`

The "startFrom" property is an optional JSON string with an enumerated value of "head", "tail", a string ISO 8601 DateTime value, or a message ID number. The default value is "tail". It sets the starting message position for a message filter in the "getMessagesFromMqTopic" action. Use it to retrieve historical messages on or after the specified date and time.

When its value is "tail", the server returns the "maxMessages" number of messages before and including the last message. The server returns messages in order of the last message toward the first.
When its value is "head", the server returns the "maxMessages" number of messages after and including the first message that has not been purged from the server. The server returns messages in order of the first message toward the first.
When it is a string ISO 8601 DateTime value, such as "2024-07-11T21:18:01.000", the server returns the "maxMessages" number of messages starting on or after that date and time. If the time component is missing, it defaults to "00:00:00.000".
When it is an ID number, such as 52233, the server returns the "maxMessages" number of messages starting on or after that ID. Messages are returned in ID ascending order. Do not embed an ID number in a string so the server can tell the difference between an ID and a DateTime.

`"topic"`

The "topic" property is a required, case-sensitive UTF-8 string containing the name of a topic. It must not contain MQTT wildcard characters, #, or +. It may contain multiple / characters, which segment a topic into a set of positional properties.

"getMessagesFromMqTopic" (MQ API)

Use the "getMessagesFromMqTopic" action to retrieve messages published on a specific topic

Tip

Tip

Request examples

Minimal

Maximal

Retrieve the most recently sent message to topic1 with all metadata and the message payload formatted as JSON

Retrieve the last 10 messages sent to topic1 with the message payload formatted as base64 and no metadata

Retrieve the last message sent to all topics, format the message payload as hexadecimal, and include all MQTT properties

Retrieve the first 50 non-purged messages sent to topic1 with the message payload formatted as JSON

Retrieve the next 50 non-purged messages sent to topic1

Retrieve 5000 messages sent to topic1 on or after "2024-07-11T21:18:01.000"

Response example

Use the "getMessagesFromMqTopic" action to retrieve messages published on a specific topic

"params"

"binaryFormat"

Note

Note

"expiredMessagesFilter"

"includeMessageMetadata"

"includeMqttProperties"

"includeOtherFields"

"includeTransformFields"

"maxMessages"

"startFrom"

"topic"

`"getMessagesFromMqTopic"` (MQ API)

Use the `"getMessagesFromMqTopic"` action to retrieve messages published on a specific topic

Retrieve 5000 messages sent to topic1 on or after `"2024-07-11T21:18:01.000"`

Use the `"getMessagesFromMqTopic"` action to retrieve messages published on a specific topic

`"params"`

`"binaryFormat"`

`"expiredMessagesFilter"`

`"includeMessageMetadata"`

`"includeMqttProperties"`

`"includeOtherFields"`

`"includeTransformFields"`

`"maxMessages"`

`"startFrom"`

`"topic"`