RabbitMQ Integration

RabbitMQ Integration connects ThingsBoard to a RabbitMQ broker over AMQP. ThingsBoard consumes messages from one or more queues, decodes them via an uplink converter, and stores the resulting telemetry and attributes. An optional downlink converter lets ThingsBoard publish messages back to a RabbitMQ exchange in response to rule engine events.

Before creating the integration, prepare the following:

• A running RabbitMQ broker (local installation or a managed service such as CloudAMQP).
• A queue to consume from. The queue must exist before the integration is created.
• Credentials (username and password) with permission to read from the queue.

CloudAMQP provides a fully managed RabbitMQ service with a free Little Lemur plan (100 queues, 1 million messages/month, 20 connections) — enough to test the integration without installing anything locally.

1. Create an instance.
   Sign up at customer.cloudamqp.com/signup. In the console, click Create New Instance, choose the Little Lemur (Free) plan, select a region close to your ThingsBoard deployment, and click Create instance.

2. Copy the connection details.
   Open the instance and go to its Overview page. Copy the following values — you will need them in the ThingsBoard Connection step:

   • Host — the hostname portion of the AMQP URL (e.g. crow.rmq.cloudamqp.com).
   • User & Vhost — the username and virtual host (both are the same string, e.g. abcdefgh).
   • Password — click Show to reveal it.

   The full AMQP URL is shown for reference: amqps://user:password@host/vhost.

3. Create the uplink queue.
   Click the green RabbitMQ Manager button on the Overview page to open the Management UI (hosted at https://<host>/).
   Go to Queues and Streams, click Add a new queue, set Durability to Durable, enter tb-uplink as the name, and click Add queue.

CloudAMQP console — create a new instance and select the Little Lemur (Free) plan. Instance Overview — copy the Host, User & Vhost, and Password values for the ThingsBoard Connection step. Management UI → Queues and Streams — add queue tb-uplink with Durability set to Durable. Queue tb-uplink created and running — ready to receive messages.

The uplink converter decodes incoming RabbitMQ messages and maps them to the ThingsBoard data model. RabbitMQ uses a generic uplink converter.

The decoder function receives:

• payload — raw AMQP message body as a UTF-8 byte array
• metadata — integrationName and any key-value pairs configured in the integration’s Metadata settings

Sample payload:

{
  "deviceName": "Sensor A1",
  "deviceType": "thermostat",
  "temperature": 23.5,
  "humidity": 60
}

1. Go to Integrations center ⇾ Data converters.
2. Click + Add data converter ⇾ Create new converter.
3. Set Converter type to Uplink (default).
4. Select integration type from the dropdown — RabbitMQ.
5. Enter a converter name: RabbitMQ Uplink Converter.
6. Paste the decoder function from the tab below.

The decoder function used in this tutorial:

// Decode an uplink message from a buffer
// payload - array of bytes
// metadata - key/value object

/** Decoder **/

// decode payload to JSON
var data = decodeToJson(payload);

// --- Device name and type ---
var deviceName = data.deviceName != null ? data.deviceName : 'Unknown Device';
var deviceType = data.deviceType != null ? data.deviceType : 'default';
// var customerName = 'Customer C';
// var groupName = 'thermostat devices';

// use assetName and assetType instead of deviceName and deviceType
// to automatically create assets instead of devices.
// var assetName = 'Asset A';
// var assetType = 'building';

// --- Timestamp parsing ---
var timestamp = -1;
if (data.ts != null) {
  timestamp = data.ts;
} else if (data.timestamp != null) {
  timestamp = new Date(data.timestamp).getTime();
}
if (timestamp == -1) {
  timestamp = Date.now();
}

// --- Telemetry and attributes ---
var telemetry = {};
var attributes = {
  model: 'Model A',
  serialNumber: 'SN111',
  integrationName: metadata['integrationName'],
};

// Keys to exclude from telemetry (already used or non-telemetry fields)
var excludeFromTelemetryList = ["deviceName", "deviceType", "ts", "timestamp"];

// Parse all remaining JSON fields as telemetry
telemetry.putAll(toFlatMap(data, excludeFromTelemetryList, true));

// Result object with device attributes/telemetry data
var result = {
  deviceName: deviceName,
  deviceType: deviceType,
  // customerName: customerName,
  // groupName: groupName,
  // assetName: assetName,
  // assetType: assetType,
  attributes: attributes,
  telemetry: {
    ts: timestamp,
    values: telemetry
  }
};

/** Helper functions 'decodeToString', 'decodeToJson' and 'toFlatMap' are already built-in **/

return result;

// Decode an uplink message from a buffer
// payload - array of bytes
// metadata - key/value object

/** Decoder **/

// decode payload to JSON
var data = decodeToJson(payload);

// --- Device name and type ---
var deviceName = data.deviceName != null ? data.deviceName : 'Unknown Device';
var deviceType = data.deviceType != null ? data.deviceType : 'default';
// var customerName = 'Customer C';
// var groupName = 'thermostat devices';

// use assetName and assetType instead of deviceName and deviceType
// to automatically create assets instead of devices.
// var assetName = 'Asset A';
// var assetType = 'building';

// --- Timestamp parsing ---
var timestamp = -1;
if (data.ts != null) {
  timestamp = data.ts;
} else if (data.timestamp != null) {
  timestamp = new Date(data.timestamp).getTime();
}
if (timestamp == -1) {
  timestamp = Date.now();
}

// --- Telemetry and attributes ---
var telemetry = {};
var attributes = {
  model: 'Model A',
  serialNumber: 'SN111',
  integrationName: metadata['integrationName'],
};

// Keys to exclude from telemetry (already used or non-telemetry fields)
var excludeFromTelemetryList = ["deviceName", "deviceType", "ts", "timestamp"];

// Parse all remaining JSON fields as telemetry
toFlatMap(data, telemetry, excludeFromTelemetryList);

// Result object with device attributes/telemetry data
var result = {
  deviceName: deviceName,
  deviceType: deviceType,
  // customerName: customerName,
  // groupName: groupName,
  // assetName: assetName,
  // assetType: assetType,
  attributes: attributes,
  telemetry: {
    ts: timestamp,
    values: telemetry
  }
};

/** Helper functions **/

function decodeToString(payload) {
  return String.fromCharCode.apply(String, payload);
}

function decodeToJson(payload) {
  var str = decodeToString(payload);
  return JSON.parse(str);
}

function toFlatMap(obj, result, excludeList, prefix) {
  prefix = prefix || '';
  for (var key in obj) {
    if (excludeList.indexOf(key) !== -1) continue;
    var value = obj[key];
    var fullKey = prefix ? prefix + '.' + key : key;
    if (typeof value === 'object' && value !== null && !Array.isArray(value)) {
      toFlatMap(value, result, excludeList, fullKey);
    } else {
      result[fullKey] = value;
    }
  }
}

return result;

7. Optionally, click Test payload decoder to validate.
8. Click Add.

To adapt this converter to your device:

• Different device name / type fields — replace data.deviceName and data.deviceType with the field names your device actually sends.
• Timestamp — if your payload includes a Unix millisecond timestamp, name the field ts; for an ISO 8601 string, use timestamp. Without either, the converter falls back to server receive time.
• Telemetry fields — all JSON fields not listed in excludeFromTelemetryList are mapped to telemetry automatically via toFlatMap. Add field names to the exclude list to keep them out of telemetry.
• Static attributes — replace model and serialNumber with actual device properties, or remove them if not needed.
• Assets instead of devices — uncomment assetName / assetType and comment out deviceName / deviceType.
• Customer or group assignment — uncomment customerName or groupName and set the appropriate values.

The downlink converter encodes Rule Engine messages into AMQP payloads. Skip this step if you only need uplink.

The output must include:

• contentType — TEXT, JSON, or BINARY
• data — the message body to publish
• metadata — optional AMQP message headers as key-value pairs

1. Go to Integrations center ⇾ Data converters.
2. Click + Add data converter ⇾ Create new converter.
3. Set Converter type to Downlink (default).
4. Select integration type from the dropdown — RabbitMQ.
5. Enter a converter name: RabbitMQ Downlink Converter.
6. Paste the decoder function from the tab below.

The encoder function used in this tutorial:

var result = {
  contentType: 'JSON',
  data: JSON.stringify(msg),
  metadata: {
    deviceName: metadata.deviceName
  }
};

return result;

var result = {
  contentType: 'JSON',
  data: JSON.stringify(msg),
  metadata: {
    deviceName: metadata.deviceName
  }
};

return result;

7. Optionally, click Test encoder function to validate.
8. Click Add.

To adapt this converter:

• Specific fields only — replace JSON.stringify(msg) with a selective object to send only the relevant command fields.
• Binary payload — set contentType to BINARY and encode data accordingly.

1. Go to Integrations center ⇾ Integrations and click + Add integration.
2. Basic settings:
   • Set Integration type to RabbitMQ.
   • Enable integration and Allow create devices or assets are on by default.
   • A 15 min badge is shown next to the name — this indicates that full debug events will be captured for the first 15 minutes after the integration is saved. No action is needed here.
   • Click Next.
3. Uplink data converter:
   • Click Select existing and choose the previously created RabbitMQ Uplink Converter from the list.
   • Click Next.
4. Downlink data converter:
   • Click Select existing and choose the previously created RabbitMQ Downlink Converter from the list.
   • Click Skip if you do not need downlink — it can be added later by editing the integration.
   • Click Next.
5. Connection:
   • Host — hostname of your RabbitMQ broker (e.g. hawk-01.rmq.cloudamqp.com).
   • Port — 5672 for plain AMQP, 5671 for AMQP over TLS.
   • Queue — add the queue to consume from (e.g. tb-uplink). Set Durable, Exclusive, and Auto delete to match the flags used when the queue was declared.
   • Expand Advanced settings and fill in Username, Password, and Virtual host. See Connection settings for details.
   • Click Next.
6. Check connection (optional):
   • Click Check connection to verify the broker is reachable. A green Connected confirmation appears on success.
   • Click Add to save the integration.

Basic settings — set Integration type to RabbitMQ, enter a name. The "15 min" badge indicates debug mode is active for the first 15 minutes after saving. Uplink data converter — click Select existing and choose the RabbitMQ Uplink Converter, then click Next. Downlink data converter — select an existing converter, or click Skip to omit downlink (can be configured later). Connection — enter the broker Host and Port, add the queue name with Durable enabled, then expand Advanced settings to fill in credentials and Virtual host. Advanced settings — on CloudAMQP, Virtual host equals the Username. Leave Downlink topic and Exchange name empty if downlink is not used. Check connection — a green "Connected" status confirms ThingsBoard reached the broker. Click Add to save the integration.

Host

Hostname or IP address of the RabbitMQ broker (e.g. localhost, rabbitmq.example.com).

Port

AMQP port. Default: 5672. Use 5671 for AMQP over TLS.

Queue

List of queues ThingsBoard consumes messages from. For each queue:

Advanced settings

Execute remotely

When enabled, ThingsBoard generates an Integration key and Integration secret that allow the integration to run as a separate process outside the ThingsBoard cluster — useful when the broker is only reachable from a restricted network. See Remote Integration.

Publish a test message to the queue using the RabbitMQ Management UI, the CloudAMQP HTTP API, or the rabbitmqadmin CLI tool.

• Management UI
• CloudAMQP HTTP API
• CLI (rabbitmqadmin)

1. Open the Management UI, go to Queues and Streams, and click the tb-uplink queue.
2. Expand Publish message.
3. Set Content type to application/json, paste the JSON payload into the Payload field, and click Publish message.

{
  "deviceName": "Sensor A1",
  "deviceType": "thermostat",
  "temperature": 23.5,
  "humidity": 60
}

Open queue tb-uplink → Publish message, paste the JSON payload, and click Publish message. A "Message published." banner confirms the message was delivered to the queue.

Check device telemetry — go to Entities ⇾ Devices. The device Sensor A1 is automatically provisioned on the first message. Open it and check the Latest telemetry tab — temperature and humidity should reflect the published values.

Entities → Devices: device Sensor A1 provisioned automatically. Latest telemetry shows humidity = 60 and temperature = 23.5. Attributes tab — client attributes set by the converter: integrationName, model, and serialNumber.

Integration events — go to Integrations center ⇾ Integrations, open the RabbitMQ integration, and check the Events tab. Click … in the Message column to inspect the raw payload consumed from the queue.

Integration Events tab — one Uplink event with Status OK. The Debug filter shows events from the last 15 minutes. Click "…" in the Message column to inspect the raw JSON payload consumed from the queue.

Converter events — go to Integrations center ⇾ Data converters, click the RabbitMQ Uplink Converter, and open its Events tab. Click … in the respective column to inspect each field:

• In — the raw payload passed to the converter.
• Out — the decoded result: deviceName, deviceType, attributes (model, serialNumber, integrationName), and telemetry (temperature, humidity).
• Metadata — integrationName injected by the integration.

Integrations center → Data converters: open the RabbitMQ Uplink Converter and navigate to the Events tab. "In" — the raw JSON payload received from the queue and passed to the decoder. "Out" — the decoded converter output: deviceName, deviceType, attributes, and telemetry values. "Metadata" — integrationName injected by the integration.

In the CloudAMQP Management UI, create a dedicated queue for outgoing messages:

1. Go to Queues and Streams and expand Add a new queue.
2. Keep Type as Default for virtual host and Durability as Durable.
3. Enter tb-downlink as the Name and click Add queue.

Add a new queue — enter tb-downlink as the Name, keep Durability set to Durable, and click Add queue. Both queues are now listed: tb-uplink for incoming messages and tb-downlink for outgoing commands.

Open the RabbitMQ integration and click the edit icon (pencil). Verify that a Downlink data converter is attached, then expand Advanced settings and set:

• Downlink topic — tb-downlink (used as the routing key when publishing downlink messages).
• Exchange name — leave empty to use the default exchange.

Click Apply changes.

Integration Details — both uplink and downlink converters must be attached before testing downlink. Advanced settings — set Downlink topic to tb-downlink and leave Exchange name empty to use the default exchange.

1. Go to Rule chains and open the Root Rule Chain. Click the edit icon.
2. Find Integration Downlink in the node panel and drag it onto the canvas.
3. Enter a name (e.g. RabbitMQ Integration), select RabbitMQ integration, and click Add.
4. Connect the Message Type Switch node to the new node via the Post attributes / Attributes Updated relation.
5. Click Apply changes.

Find Integration Downlink in the node panel and drag it onto the canvas. Enter a name, select RabbitMQ integration from the dropdown, and click Add. Connect the Message Type Switch node to the Integration Downlink node via the Post attributes / Attributes Updated relation, then click Apply changes.

Trigger a downlink by adding a shared attribute to device Sensor A1:

1. Go to Entities ⇾ Devices and open Sensor A1.
2. Navigate to the Attributes tab, switch to Shared attributes, and click +.
3. Enter key powerState and value on, then click Add.

Adding the attribute fires the rule chain, which forwards the message through the Integration Downlink node to the RabbitMQ integration. The downlink converter encodes the message and ThingsBoard publishes it to the tb-downlink queue via the default exchange.

In the Management UI at https://hawk.rmq.cloudamqp.com/, go to Queues and Streams and open the tb-downlink queue. Scroll to Get messages, set Ack mode to Nack message requeue true (peek without consuming), and click Get Message(s).

The result shows:

• Exchange — (AMQP default)
• Routing Key — tb-downlink
• Payload — the downlink converter output, e.g. {"powerState":"on"}

Integration status is not Active

1. Confirm the Host and Port are reachable from the ThingsBoard server.
2. Verify the Username and Password in Advanced settings are correct.
3. Check that the Virtual host matches the one on the broker (default /).
4. Ensure the queue exists on the broker before the integration tries to connect.

Messages are not appearing in ThingsBoard

1. Check the integration Events tab for channel-level errors — a flag mismatch is shown there.
2. Verify the Durable, Exclusive, and Auto delete flags match the queue’s original declaration.
3. Confirm the publisher is routing messages to the correct queue (check bindings in the Management UI).
4. Use the Converter Events tab to inspect the raw In payload — verify the uplink converter handles the actual message format.

Downlink messages are not published

1. Confirm the downlink converter is attached to the integration (edit the integration and check step 4).
2. Verify the Integration Downlink rule node is correctly connected in the rule chain.
3. Check the rule node’s Events tab for processing errors.

• Integration Overview
• Integration types
• Uplink data converter
• Downlink data converter
• Remote Integration
• Rule Engine
• ThingsBoard Expression Language (TBEL)