Sources and data import

Why this matters

Not every device connects to your EMQX. You may have a Zigbee gateway or a cloud platform that already publishes its data to another MQTT broker. To get those external messages into your EMQX, and from there into a rule or into Home Assistant, you need a way in — and that is a Source.

The Sink in Chapter 11 sends data out; this chapter goes the other way, into data import: an MQTT Source collects data from a remote broker, and a republish action (or a rule) writes those external messages back onto a topic your local EMQX can use.

Core concepts

Data bridging between MQTT brokers (many people call it an MQTT broker bridge, and the EMQX 5.x documentation calls it MQTT Broker Data Integration) lets EMQX connect to another MQTT service as a client and exchange messages in both directions:

• Egress (Sink): publishes messages from your local EMQX to a chosen topic on the remote broker.
• Ingress (Source): subscribes to a topic on the remote broker and feeds what it receives back into your local EMQX.

Both share the same Connector (Chapter 11). One connection can carry several bridge rules, each with its own topic mapping and transformation. For a local client to read what a Source brings in, you normally add a republish action that writes it back locally.

Terms at a glance

Hands-on

Following the official tutorial: bridge what a remote broker (broker.emqx.io) receives on f/# onto your local sub/#.

1. Create the MQTT Broker Connector

   In the Dashboard go to Integration → Connectors → Create and pick MQTT Broker. Name it my_mqtt_bridge, set the server to broker.emqx.io:1883 (fill in the credentials if the remote broker requires authentication), then click Create.

2. Create the rule and add the MQTT Source

   Go to Integration → Rules → Create. On the "Data Inputs" tab, delete the default Message input, click Add Input, pick MQTT Broker, choose the Connector, then set the subscribe Topic ($share/1/f/#, for example) and the QoS.

3. Check the rule SQL

   Once the Source is added, the rule SQL becomes SELECT * FROM "$bridges/mqtt:my_source" on its own, which says the data source is this MQTT broker bridge.

4. Add the Republish action

   Switch to the "Action Outputs" tab, click + Add Action and pick Republish. Set Topic to sub/${topic}, QoS to ${qos} and Payload to ${payload}.

5. Create it and verify

   Click Create to finish the rule. Subscribe to sub/# locally, publish a message to f/1 on the remote broker, and it should arrive locally on sub/f/1.

MQTT Source and the $bridges syntax

Two things to watch when you create an MQTT Source:

• The subscribe topic accepts the + and # wildcards, so one Source can pick up a whole family of remote topics.
• Use a shared subscription with a cluster or a connection pool: when EMQX runs as a cluster, or the Connector has a connection pool open, several clients subscribe to the same topic at once and each of them receives the message. The official advice is to spread that load with $share/<group>/topic (for example $share/1/f/#).

The fields a Source brings in, for use in the rule SQL, are:

A rule uses $bridges/mqtt:<name> as its data source, where <name> is the name of the Connector/Source.

Bridging across brokers and writing back to EMQX

The egress side is simple: create an MQTT Sink (egress) that publishes a local topic to the remote broker. A rule of SELECT * FROM "t/#" pointed at the target pub/${topic}, for example, forwards your local t/1 to pub/t/1 on the remote broker. On the way out, QoS and retain can carry over from the original message through the ${qos} and ${flags.retain} placeholders.

Ingress write-back is what a Source is for: it pulls the messages it subscribed to on the remote broker into a rule, but it does not publish them locally by itself — you have to add a Republish action to write them back onto a local topic. The official example uses sub/${topic}, so a remote f/1 lands on your sub/f/1, and a local client that subscribes to sub/# receives it.

To declare all of this in a config file instead, create the matching ingress (Source) and egress (Sink) bridges in the rule_engine and connector/bridges sections, mirroring the rule's SQL. On a cluster or across several nodes a fixed MQTT client ID would collide between them, which is why a fixed client ID is not supported here.

Troubleshooting

• The Source receives nothing from the remote broker: check that the Connector state is Connected, that the subscribe Topic and QoS are right, that the wildcards are correct, and that something really is publishing on the remote broker.
• Duplicate data with a cluster or a connection pool: switch to a shared subscription, $share/<group>/topic, so several bridge clients do not each receive the same message.
• The written-back messages never arrive locally: a Source only brings the message in; it takes a separate Republish action to actually write it back onto a local topic. Check that the action was added and the rule created.
• A SQL field comes back empty: check that the rule's FROM is the right $bridges/mqtt:<name>, and that the name you put in matches the Source's name.

FAQ

Official sources

• EMQX 5.8 MQTT Broker Bridge (pinned docs branch)
• EMQX 5.8 Connector (pinned docs branch)
• EMQX 5.8 Data Sources and Fields (pinned docs branch)