RabbitMQ Integration

How to integrate RabbitMQ with Connectware through the MQTT plugin, including topic-to-routing-key translation, publishing shop floor data to AMQP consumers, and receiving messages in return.

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 with sufficient user permissions.

• Basic knowledge of MQTT and the Connectware services concept (for example, service commissioning files, connections, endpoints, and mappings).

Connectware and RabbitMQ Integration

RabbitMQ is a message broker whose native protocol is the Advanced Message Queuing Protocol (AMQP). The RabbitMQ MQTT plugin ships with every RabbitMQ distribution and adds an MQTT listener to the broker. Connectware connects to this listener through the MQTT connector, 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):

• 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:

rabbitmq-plugins enable rabbitmq_mqtt

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).

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

rabbitmq-plugins enable rabbitmq_management

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

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

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).

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.

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

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.

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

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.

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.

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

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).

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.

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

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:

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

1. 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.

2. 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 to check that the message arrives on enterprise/backend/erp/orders.

Service Commissioning File Example

rabbitmq-example.yml

3KB

Open

rabbitmq-example.yml

---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: [e-mail]
# ----------------------------------------------------------------------------
# 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