RabbitMQ integration

TBMQ RabbitMQ Integration publishes MQTT messages and client lifecycle events to an exchange on a RabbitMQ broker of your own, over AMQP 0-9-1. You choose the exchange and how each message’s routing key is formed — derived from the MQTT topic, or a fixed one — and RabbitMQ’s own bindings decide which queues the message lands in. This is useful for the following scenarios:

• Fan-out to existing consumers — handing device data to applications, workers and services that already consume from RabbitMQ, without teaching them MQTT.
• Topic-based routing — publishing to a topic exchange where each consumer binds the slice of the MQTT topic tree it cares about, e.g. sensors.*.temp.
• Work queues — feeding a durable queue that a pool of workers drains at its own pace, with the broker holding the backlog.
• Bridging to other protocols — reaching AMQP, STOMP and other RabbitMQ-side consumers through one broker instead of adding a second integration per protocol.

1. A client publishes an MQTT message to a topic matching the integration’s topic filters.
2. TBMQ receives the message and forwards it to the TBMQ Integration Executor over Kafka.
3. The Integration Executor resolves the routing key, builds the body, and publishes one AMQP message to the configured exchange.
4. RabbitMQ routes it to every queue bound to that exchange with a matching binding key, and returns a publisher confirm. The message is acknowledged only after that confirm arrives.

• A running TBMQ instance with the TBMQ Integration Executor service running. Without a reachable executor, saving an integration fails with a validation timeout.
• A RabbitMQ broker reachable from the Integration Executor, speaking AMQP 0-9-1.
• The exchange must already exist, and queues must already be bound to it. The integration declares nothing — no exchange, no queue, no binding.
• A client capable of publishing MQTT messages (e.g., TBMQ WebSocket Client).
• The broker’s management plugin (rabbitmq-plugins enable rabbitmq_management) only if you want to follow Send an uplink message — it serves the management UI on port 15672, where that walkthrough declares the queue and reads the message back. The integration itself does not need it.

1. Go to the Integrations page and click the “+” button.
2. Select RabbitMQ as the integration type and click Next.
3. On the Filters and events step, keep the default Topic filters entry tbmq/#, leave Lifecycle events empty, and click Next.
4. On the Configuration step, fill in Host, Port and Virtual host. If your broker requires authentication, set Credentials to Basic and enter the username and password.
5. Under Routing, keep the pre-filled Exchange amq.topic and leave Use message topic as routing key on, or switch it off and enter a static Routing key.
6. Click Check connection to confirm the broker is reachable and the exchange exists, then click Add to save the integration.

Topic filters define MQTT-based subscriptions that trigger the integration. When TBMQ receives a message matching a configured topic filter, the integration resolves its routing key and publishes it to the exchange.

Topic filters are optional if you select at least one lifecycle event type instead — an integration only has to have one of the two. The form enforces that from the other side too: the last topic filter cannot be removed until an event type is selected, so to make an integration events-only, pick the event type first and then delete tbmq/#. Without topic filters the Use message topic as routing key toggle and the Routing key field disappear — there is no message to route, and lifecycle events never use them. Exchange stays, because events are published to it too, but it is no longer required: left empty, events go through the default exchange to the queue named by the Events routing key.

For example, with the topic filter tbmq/devices/+/status, any of the following messages will trigger the integration:

tbmq/devices/device-01/status
tbmq/devices/gateway-01/status

AMQP always authenticates, so every connection presents something. Which SASL mechanism is used follows from the credentials type:

PEM certificate has two requirements the form enforces on save:

• Enable SSL is forced on and its toggle is hidden — SASL EXTERNAL over a plaintext connection proves nothing.
• Both a Certificate file and a Private key file are required. A CA certificate on its own is rejected, because it would connect as no one and the broker would answer with an opaque ACCESS_REFUSED.

On the broker side this needs the rabbitmq_auth_mechanism_ssl plugin enabled, and a user whose name matches the certificate. Which field of the certificate becomes that name is the broker’s own ssl_cert_login_from setting: distinguished_name by default, so the user to create is the full DN (CN=tbmq-ie,O=Acme), not the Common Name alone. Set it to common_name if you would rather name users after the CN.

Enable SSL connects with TLS instead of plain TCP. The server certificate chain and its hostname are always verified — there is no switch to weaken that.

The CA certificate file belongs to the PEM certificate credentials type only. With Basic or Anonymous credentials the form has no CA field, so the broker’s certificate has to validate against the executor’s default truststore. For a broker fronted by a private CA, either add that CA to the JVM truststore of the tbmq-ie deployment or use PEM certificate credentials.

Every PEM is parsed when the integration is saved, so a malformed certificate or a wrong key password is reported there rather than leaving the integration to fail on start.

amq.topic is a topic exchange that RabbitMQ predeclares on every vhost and refuses to delete, which makes it a working starting point. A custom exchange is the case to watch — it has to exist on the broker before the integration can publish to it, and Check connection is what surfaces a missing or misspelled one.

The confirm timeout is capped at 25 seconds so that no message can outlive the executor’s message pack timeout (TB_IE_MSG_PACK_PROCESSING_TIMEOUT, 30 000 ms by default) without a callback. If you lower that timeout below 25 000 ms, lower this value with it — the cap alone no longer holds the guarantee.

With Use message topic as routing key on, each message’s routing key is its MQTT topic with / replaced by .:

sensors/room1/temp   →   sensors.room1.temp

That is the whole transformation. Dots already in the topic are kept, so a/b.c and a/b/c both map to a.b.c — two MQTT topics that collapse onto one routing key.

This mode requires a named exchange, normally a topic exchange, where consumers bind the slices they want (sensors.#, sensors.*.temp). The default exchange is rejected for it on save, because it routes by queue name and a topic-derived key cannot be relied on to name one.

A key is limited to 255 bytes — the AMQP 0-9-1 short-string limit. A static routing key is checked against it when the integration is saved; a topic-derived one can only be checked when the message arrives, so an over-long topic fails that message and the acknowledgment strategy decides what happens next.

With the toggle off, every message carries the same Routing key. Where that message ends up is entirely RabbitMQ’s call: a direct exchange matches the key exactly, a fanout exchange ignores it, a topic exchange matches it against the binding patterns.

Leaving Exchange empty selects the AMQP default exchange. It ignores bindings and delivers to the queue whose name equals the routing key, which makes it the shortest path to a single known queue — set a static routing key to the queue’s name.

Because there are no bindings to get wrong, this is also the one mode where Check connection can verify the destination itself: it passively declares the queue instead of the exchange.

With Send only message payload off (the default), the body is the message envelope as JSON:

{
  "payload": "eyJ0ZW1wZXJhdHVyZSI6MjV9",
  "topicName": "tbmq/rabbitmq-integration",
  "clientId": "tbmq_df52bNUQ",
  "eventType": "PUBLISH_MSG",
  "qos": 1,
  "retain": false,
  "tbmqIeNode": "tbmq-ie-0",
  "tbmqNode": "tbmq-node-0",
  "ts": 1742554969254,
  "props": {},
  "metadata": {
    "integrationName": "RabbitMQ integration"
  }
}

payload holds the MQTT payload Base64-encoded — the envelope carries it as raw bytes, whatever it contains, so text and binary payloads are encoded alike. eyJ0ZW1wZXJhdHVyZSI6MjV9 above decodes to {"temperature":25}. clientCertCn is present only for a client that authenticated with mTLS.

With Send only message payload on, the body is the raw payload bytes, published byte for byte — binary safe, never re-encoded and never Base64. Nothing else about the message reaches RabbitMQ, so anything a consumer needs to tell messages apart has to come from the routing key or from a static header.

Every published message carries the same fixed set of properties, built once from the configuration:

Nothing is derived per message — there is no expiration, priority, correlation-id or message-id, and no MQTT property is mapped onto an AMQP one. Everything about a message that a consumer may need is in the body or in the routing key.

Besides the messages matched by topic filters, this integration can deliver client lifecycle events — a client connecting, disconnecting, changing its subscriptions, or failing authentication or authorization.

Pick the event types on the Filters and events step while creating the integration, or add them later:

1. Open the integration on the Integrations page and click the Toggle edit mode button (pencil icon).
2. In the Lifecycle events section, click the Event types field to list the available event types, then select the ones you need: Client connected, Client disconnected, Client subscribed, Client unsubscribed, Client authentication failed, Client authorization failed, Client connection failed. Each one is added as a chip; remove it with its x.
3. Fill in the Events routing key that appears below.
4. Click Apply changes.

Events go to the same Exchange with the same AMQP properties as messages, but always under the static Events routing key — an event has no MQTT topic to derive one from, which is why the key is required as soon as any event type is selected, and why Use message topic as routing key has no bearing on events.

The body is always the whole event envelope as JSON — its common fields plus the ones the event type adds, as Event payload describes them. Send only message payload is a message-only switch: an event carries no MQTT payload to send on its own.

Check connection does more than open a socket:

1. Connects with the configured virtual host, credentials and TLS settings, and opens a channel.
2. Passively declares the exchange, when a named exchange is configured. A passive declare asks the broker whether the exchange exists and creates nothing; an exchange that is missing closes the channel, and the failure is reported with the broker’s own reason.
3. Passively declares the queues instead, when the default exchange is used: the queue named by the Routing key while topic filters are configured, and the one named by the Events routing key while lifecycle events are selected.

Nothing is declared, published or consumed — every step above is a read.

The probe’s budget is the Connection timeout, clamped by INTEGRATIONS_INIT_CONNECTION_TIMEOUT_SEC (15 seconds by default) when that is lower, so it cannot run past the API request deadline.

Publisher confirms are always on, and the confirm is the only signal the integration reports success from: a message is acknowledged to the broker-side pack only once RabbitMQ has confirmed it.

Delivery is not exactly-once:

• A publish that fails is logged, counted, saved as an error event, and reported as a failure. It is never acknowledged as a success. What happens next is decided by the executor’s acknowledgment strategy — SKIP_ALL (the default) drops the message; RETRY_ALL redelivers it up to the configured retry limit.
• A retried message is published a second time. RabbitMQ has no deduplication, so a consumer sees both copies.

These are the failure modes, and each one fails only the messages it actually affects:

Client lifecycle events are weaker still — they are best-effort hints, and can be dropped before they ever reach the integration. See Delivery guarantees.

A RabbitMQ integration holds one AMQP connection, opened when it starts and closed when it stops. The broker sees it under the name tbmq-ie-rabbitmq-<integrationId>-<serviceId> and with the client property product = TBMQ, so it can be identified in the RabbitMQ management UI.

On that connection sit Channel pool size publisher threads, each owning one channel for the life of the integration. Publishing is a single non-blocking frame write and confirms arrive asynchronously on the client’s own connection thread, so a slow broker costs latency, not threads.

Recovery is handled by the AMQP client: with Automatic connection recovery on, a lost connection is re-established in the background. Topology recovery is deliberately off, since the integration declares no exchanges, queues or bindings to restore; channels are simply reopened on demand.

Which executor pays all of this depends on partition ownership — each integration is owned by exactly one executor instance at a time.

These are the integration’s own events, recorded inside TBMQ for debugging and troubleshooting. They are not the client lifecycle events described above: those are about MQTT clients and are published to RabbitMQ, while these describe the integration itself and never leave TBMQ.

• Lifecycle Events — logs events such as Created, Updated, Started, Stopped, Failed. A connection lost to the broker is reported as Stopped, and its automatic recovery as Started, so an outage is visible here as a pair.
• Statistics — the integration’s own counters: messages processed and errors occurred.
• Errors — captures failures such as an unreachable broker, a nacked or unconfirmed message, a channel the broker closed, or a blocked connection. Each entry carries the reason as the broker or the client reported it.

The integration publishes to an exchange, and RabbitMQ silently drops a message no queue is bound for — so bind one first. With the defaults, the amq.topic exchange and the MQTT topic as the routing key, open the management UI at http://<broker-host>:15672 and:

1. On the Queues and Streams tab, expand Add a new queue. Set Type to Classic and Name to tbmq-demo, leave Durability at Durable, and click Add queue.
2. On the Exchanges tab, click amq.topic.
3. Expand Bindings. Under Add binding from this exchange keep To queue selected, enter tbmq-demo as the destination and tbmq.# as the Routing key, and click Bind.

Then publish a message:

1. Navigate to the WebSocket Client page.
2. Select WebSocket Default Connection (or any working connection) and click Connect. Verify the connection status shows Connected.
3. Set the Topic field to tbmq/rabbitmq-integration to match the integration’s topic filter tbmq/#.
4. Leave the pre-filled Payload {"temperature": 25} as is.
5. Click the Send icon to publish the message.

The message should now be in the queue, published under the routing key tbmq.rabbitmq-integration. Back in the management UI, open Queues and Streams, click tbmq-demo, expand Get messages and click Get Message(s) — the default Ack Mode, Nack message requeue true, leaves the message in the queue:

{"payload":"eyJ0ZW1wZXJhdHVyZSI6IDI1fQ==","topicName":"tbmq/rabbitmq-integration","clientId":"tbmq_df52bNUQ","eventType":"PUBLISH_MSG","qos":1,"retain":false,"tbmqIeNode":"tbmq-ie-0","tbmqNode":"tbmq-node-0","ts":1742554969254,"props":{},"metadata":{"integrationName":"RabbitMQ integration"}}

Note that payload is Base64-encoded: inside the envelope the MQTT payload always travels as bytes, so eyJ0ZW1wZXJhdHVyZSI6IDI1fQ== is the {"temperature": 25} that was published, and a consumer has to decode it. Turning Send only message payload on would have published {"temperature": 25} alone — byte for byte, with no envelope and no encoding.