> For the complete documentation index, see [llms.txt](https://docs.cybus.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.cybus.io/2-4-2/guides/system-connectivity/messaging-event-streaming/rabbitmq-integration.md).

# RabbitMQ Integration

This guide describes how to integrate RabbitMQ with Connectware. You enable the RabbitMQ MQTT plugin and configure a service commissioning file that publishes shop floor data to RabbitMQ and receives messages from AMQP backends in return. A complete example file is available at the end of this guide.

## Objectives

* Enabling the RabbitMQ MQTT plugin and creating a RabbitMQ user for Connectware.
* Establishing an MQTT connection between Connectware and RabbitMQ.
* Publishing shop floor data from an ISA-95 topic hierarchy to AMQP consumers.
* Receiving messages from AMQP backends in Connectware.

## Prerequisites

To follow this guide, you will need the following:

* A running instance of Cybus Connectware.
* A running RabbitMQ node or cluster, with administrative access to it, including the `rabbitmq-plugins` and `rabbitmqctl` commands. The verification steps also use the RabbitMQ management UI.
* Access to the [Admin UI](/2-4-2/access/admin-ui.md) with sufficient [user permissions](/2-4-2/access/user-management.md).
* Basic knowledge of MQTT and the Connectware [services](/2-4-2/data-flows/services.md) concept (for example, [service commissioning files](/2-4-2/data-flows/service-commissioning-files.md), [connections](/2-4-2/data-flows/service-commissioning-files/resources/cybus-connection.md), [endpoints](/2-4-2/data-flows/service-commissioning-files/resources/cybus-endpoint.md), and [mappings](/2-4-2/data-flows/service-commissioning-files/resources/cybus-mapping.md)).

## Connectware and RabbitMQ Integration

RabbitMQ is a message broker whose native protocol is the Advanced Message Queuing Protocol (AMQP). The [RabbitMQ MQTT plugin](https://www.rabbitmq.com/docs/mqtt) ships with every RabbitMQ distribution and adds an MQTT listener to the broker. Connectware connects to this listener through the [MQTT connector](/2-4-2/connectors/enterprise-connectors/mqtt.md), like to any other MQTT broker. No gateway or custom bridge is required.

The plugin is not a separate broker next to RabbitMQ. Every message that arrives over MQTT is routed through a regular AMQP topic exchange, `amq.topic` by default. This makes the plugin a bridge between the two protocol worlds: messages that Connectware publishes over MQTT reach AMQP consumers through queue bindings, and messages that AMQP backends publish to the `amq.topic` exchange reach the MQTT subscriptions of Connectware.

The RabbitMQ MQTT plugin differs from a generic MQTT broker in a few ways that matter for this integration (see the [RabbitMQ MQTT plugin documentation](https://www.rabbitmq.com/docs/mqtt)):

* The plugin supports MQTT 3.1, 3.1.1, and 5.0 with Quality of Service (QoS) levels 0 and 1. QoS 2 is not supported: RabbitMQ downgrades QoS 2 to QoS 1 for MQTT 3.1.1 clients and rejects QoS 2 publishes from MQTT 5.0 clients. Do not configure `qos: 2` on endpoints or mappings that target RabbitMQ.
* Retained messages are supported with limitations. The retained message store is node-local: retained messages are neither replicated to nor queried from other cluster nodes, and subscriptions whose topic filter contains wildcards do not receive retained messages. Do not rely on the `retain` flag when RabbitMQ runs as a cluster.
* Authentication uses regular RabbitMQ users and permissions. The default `guest` user can only connect from localhost, so Connectware needs a dedicated user.
* The dot character (`.`) is the routing key separator on the AMQP side. Avoid dots inside MQTT topic segments, otherwise the topic translates to a routing key with unintended extra levels.
* Every MQTT subscription is backed by a queue named `mqtt-subscription-<clientId>qos<level>`. If a second client connects with the same client ID, RabbitMQ closes the first connection. Choose a client ID that is unique on the broker.

The MQTT topics on the Connectware side follow an ISA-95-style equipment hierarchy (`<enterprise>/<site>/<area>/<line>/<cell>`). The mapping subscribes with wildcards across all levels, so any machine in the hierarchy is picked up without changing the integration. This guide uses `enterprise` as the root of the hierarchy. If your hierarchy uses a different root, adjust the topics in the mapping resources.

### How MQTT Topics Map to AMQP Routing Keys

The MQTT plugin routes every message through the `amq.topic` exchange. The MQTT topic becomes the AMQP routing key, with the separator translated in both directions:

| MQTT                      | AMQP                      |
| ------------------------- | ------------------------- |
| Topic separator `/`       | Routing key separator `.` |
| Single-level wildcard `+` | Binding key wildcard `*`  |
| Multi-level wildcard `#`  | Binding key wildcard `#`  |

For example, a message that Connectware publishes to the MQTT topic `enterprise/hamburg/assembly/line-1/press-01/telemetry` arrives on the `amq.topic` exchange with the routing key `enterprise.hamburg.assembly.line-1.press-01.telemetry`. An AMQP consumer that binds a queue to `amq.topic` with the binding key `enterprise.#` receives every message of the hierarchy. The binding key `enterprise.hamburg.assembly.line-1.*.telemetry` narrows this down to the telemetry of a single line.

The same translation applies in the opposite direction: an AMQP message published to `amq.topic` with the routing key `backend.erp.orders` is delivered to MQTT subscribers of the topic `backend/erp/orders`.

The exchange is configurable with the `mqtt.exchange` key in the `rabbitmq.conf` file. This guide uses the default, `amq.topic`.

### Preparing RabbitMQ

1. Enable the MQTT plugin on the RabbitMQ node:

{% code lineNumbers="true" %}

```bash
rabbitmq-plugins enable rabbitmq_mqtt
```

{% endcode %}

The plugin accepts MQTT connections on port 1883. To accept TLS connections on port 8883, configure the TLS listener and certificates first (see [RabbitMQ TLS support](https://www.rabbitmq.com/docs/ssl)).

2. Enable the management plugin. The verification steps in this guide use the management UI, which is served on port 15672:

{% code lineNumbers="true" %}

```bash
rabbitmq-plugins enable rabbitmq_management
```

{% endcode %}

3. Create a dedicated user for Connectware and grant it permissions on the virtual host:

{% code lineNumbers="true" %}

```bash
rabbitmqctl add_user connectware ${RABBITMQ_PASSWORD}
rabbitmqctl set_permissions --vhost / connectware ".*" ".*" ".*"
```

{% endcode %}

Replace `${RABBITMQ_PASSWORD}` with a password of your choice. The three `".*"` patterns grant configure, write, and read permissions on all resources of the virtual host. The MQTT plugin needs them to declare the subscription queues and to publish to the `amq.topic` exchange. In production, restrict the patterns to the queues and exchanges that the integration uses (see [RabbitMQ access control](https://www.rabbitmq.com/docs/access-control)).

MQTT connections open in the default virtual host `/`. To place the integration in a different virtual host, either set the `mqtt.vhost` key in the `rabbitmq.conf` file, or select the virtual host per connection by prefixing the username with the virtual host name and a colon, for example `production:connectware`.

### RabbitMQ Connection Parameters

The connection requires the hostname of the RabbitMQ node, the credentials of the Connectware user, and a client ID. We add them as parameters to the service commissioning file, so you can set them when you install the service.

Do not worry about copying the service commissioning file snippets together into one, the complete example file is available at the end of this guide.

* `rabbitmqHost`: The hostname or IP address of the RabbitMQ node.
* `rabbitmqUsername` and `rabbitmqPassword`: The credentials of the RabbitMQ user created for Connectware. To select a virtual host, prefix the username with the virtual host name and a colon.
* `clientId`: The MQTT client ID. RabbitMQ derives the names of the subscription queues from it, and closes an existing connection that uses the same client ID.

{% code lineNumbers="true" %}

```yaml
parameters:
  rabbitmqHost:
    description: Hostname or IP address of the RabbitMQ node
    type: string
    default: rabbitmq.example.com

  rabbitmqUsername:
    description: RabbitMQ user for the MQTT connection
    type: string
    default: connectware

  rabbitmqPassword:
    description: Password of the RabbitMQ user
    type: string

  clientId:
    description: MQTT client ID, determines the names of the subscription queues
    type: string
    default: connectware-shopfloor
```

{% endcode %}

### RabbitMQ Connection

To connect to the MQTT listener of RabbitMQ, we set up a `Cybus::Connection` resource that uses the MQTT connector with the username and password of the Connectware user.

{% code lineNumbers="true" %}

```yaml
resources:
  rabbitMqConnection:
    type: Cybus::Connection
    properties:
      protocol: Mqtt
      connection:
        host: !ref rabbitmqHost
        port: 1883
        scheme: mqtt
        username: !ref rabbitmqUsername
        password: !ref rabbitmqPassword
        clientId: !ref clientId
        keepalive: 60
```

{% endcode %}

The MQTT connector defaults to a keep-alive interval of `0`, which disables client-side keep-alive pings. A value of 60 seconds lets both sides detect a broken connection even when no data flows.

To connect over TLS, set `scheme: mqtts` and `port: 8883`, and provide the CA certificate with the `caCert` property if the broker certificate is not signed by a public CA. For all available connection parameters, see [MQTT Connection Properties](/2-4-2/connectors/enterprise-connectors/mqtt/mqttconnection.md).

### Publishing Shop Floor Data to RabbitMQ

A `Cybus::Mapping` resource forwards telemetry from the ISA-95 topic hierarchy to RabbitMQ. The named wildcards `+site`, `+area`, `+line`, and `+cell` capture the levels of the source topic, and the publish side reuses them as `$site`, `$area`, `$line`, and `$cell`, so the topic hierarchy on the RabbitMQ side mirrors the Connectware side.

{% code lineNumbers="true" %}

```yaml
telemetryMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          topic: enterprise/+site/+area/+line/+cell/telemetry
          qos: 1
        publish:
          connection: !ref rabbitMqConnection
          topic: enterprise/$site/$area/$line/$cell/telemetry
          qos: 1
```

{% endcode %}

A message published to `enterprise/hamburg/assembly/line-1/press-01/telemetry` now arrives on the `amq.topic` exchange with the routing key `enterprise.hamburg.assembly.line-1.press-01.telemetry`. Any AMQP consumer with a matching queue binding receives it, no MQTT client library required on the backend side. The payload is forwarded unchanged. To reshape it before it reaches RabbitMQ, for example to add a timestamp or flatten a structure, add rules to the mapping (see [Rule Engine](/2-4-2/data-flows/rule-engine.md)).

QoS 1 on both sides gives at-least-once delivery from the internal broker to RabbitMQ. Because RabbitMQ does not support QoS 2, this is the highest end-to-end guarantee available through the MQTT plugin.

### Receiving Data from AMQP Backends

For the opposite direction, a `Cybus::Endpoint` resource subscribes to a topic that an AMQP backend publishes to, and a mapping republishes every message to the Connectware topic hierarchy, where any other Connectware service can pick it up, for example to write an order number to a PLC.

{% code lineNumbers="true" %}

```yaml
ordersEndpoint:
  type: Cybus::Endpoint
  properties:
    protocol: Mqtt
    connection: !ref rabbitMqConnection
    subscribe:
      topic: backend/erp/orders
      qos: 1

ordersMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          endpoint: !ref ordersEndpoint
        publish:
          topic: enterprise/backend/erp/orders
```

{% endcode %}

Any AMQP backend that publishes to the `amq.topic` exchange with the routing key `backend.erp.orders` now reaches the shop floor. Connectware republishes the messages to `enterprise/backend/erp/orders`.

On the RabbitMQ side, this subscription appears as a queue named `mqtt-subscription-connectware-shopfloorqos1`, bound to the `amq.topic` exchange with the binding key `backend.erp.orders`. The MQTT plugin creates and binds this queue automatically.

## Verifying the Integration

1. Install the service and set the parameters with the values of your RabbitMQ node.
2. Check that the connection is in the **Connected** state on the service details page in the Admin UI. If the credentials are wrong or the user lacks permissions, the connection does not reach the connected state, and the RabbitMQ log shows the failed connection attempt.
3. Open the RabbitMQ management UI on `http://${RABBITMQ_HOST}:15672` and log in with an administrative user. The **Connections** tab shows the Connectware connection with the protocol `MQTT`, and the queues overview shows the subscription queue `mqtt-subscription-connectware-shopfloorqos1`.
4. Create a test queue named `telemetry-test` and bind it to the `amq.topic` exchange with the binding key `enterprise.#`. You can do this on the queue detail page of the management UI, or with `rabbitmqadmin`:

{% code lineNumbers="true" %}

```bash
rabbitmqadmin queues declare --name telemetry-test --durable true
rabbitmqadmin bindings declare --source amq.topic --destination-type queue --destination telemetry-test --routing-key "enterprise.#"
```

{% endcode %}

5. Publish a test message to `enterprise/hamburg/assembly/line-1/press-01/telemetry`, for example with an MQTT client or the Admin UI. On the detail page of the `telemetry-test` queue, the message counter increases, and **Get messages** shows the message with the routing key `enterprise.hamburg.assembly.line-1.press-01.telemetry`.
6. For the opposite direction, open the `amq.topic` exchange in the management UI and use **Publish message** with the routing key `backend.erp.orders` and a test payload. Use the [Data Explorer](/2-4-2/monitoring/data-explorer.md) to check that the message arrives on `enterprise/backend/erp/orders`.

## Service Commissioning File Example

{% file src="/files/fAeop9nOGnwNKyyysXLY" %}

{% code title="rabbitmq-example.yml" lineNumbers="true" expandable="true" %}

```yaml
---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: support@cybus.io
# ----------------------------------------------------------------------------
# RabbitMQ Integration (Example)
# ----------------------------------------------------------------------------

description: >
  Service commissioning file for the bidirectional integration between
  Connectware and RabbitMQ through the RabbitMQ MQTT plugin (Example)

metadata:
  name: RabbitMQ Integration
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  rabbitmqHost:
    description: Hostname or IP address of the RabbitMQ node
    type: string
    default: rabbitmq.example.com

  rabbitmqUsername:
    description: RabbitMQ user for the MQTT connection
    type: string
    default: connectware

  rabbitmqPassword:
    description: Password of the RabbitMQ user
    type: string

  clientId:
    description: MQTT client ID, determines the names of the subscription queues
    type: string
    default: connectware-shopfloor

resources:
  # Connection to the MQTT listener of the RabbitMQ MQTT plugin
  rabbitMqConnection:
    type: Cybus::Connection
    properties:
      protocol: Mqtt
      connection:
        host: !ref rabbitmqHost
        port: 1883
        scheme: mqtt
        username: !ref rabbitmqUsername
        password: !ref rabbitmqPassword
        clientId: !ref clientId
        keepalive: 60

  # Forwards shop floor telemetry to RabbitMQ. The MQTT plugin publishes each
  # message to the amq.topic exchange, with the topic as the routing key
  # (slashes translated to dots).
  telemetryMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            topic: enterprise/+site/+area/+line/+cell/telemetry
            qos: 1
          publish:
            connection: !ref rabbitMqConnection
            topic: enterprise/$site/$area/$line/$cell/telemetry
            qos: 1

  # Receives messages that AMQP backends publish to the amq.topic exchange
  # with the routing key backend.erp.orders
  ordersEndpoint:
    type: Cybus::Endpoint
    properties:
      protocol: Mqtt
      connection: !ref rabbitMqConnection
      subscribe:
        topic: backend/erp/orders
        qos: 1

  ordersMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref ordersEndpoint
          publish:
            topic: enterprise/backend/erp/orders
```

{% endcode %}


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.cybus.io/2-4-2/guides/system-connectivity/messaging-event-streaming/rabbitmq-integration.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.