Scenario: Ingesting Live Data from an External Broker

Subscribe to an external MQTT broker, transform gateway and HVAC messages in Node-RED, and ingest them into OCI IoT.

Use this scenario when devices or gateways already publish telemetry to an external MQTT broker. The flow is MQTT-IN -> Function -> MQTT-OUT, with an optional Debug branch.

Tasks

Create the Flow Runtime
Import the Example Flow or Configure It Manually
Publish Test Messages

Prerequisites

Complete the common scenario setup before starting. The example receives data from test.mosquitto.org:8883, converts public topics to the adapter topics, and publishes to the OCI IoT device host.
A private subnet with a route through a NAT gateway for 0.0.0.0/0, or another approved route that lets the runtime reach the external broker.
Network security rules that permit outbound TLS connections to the broker on TCP port 8883 and to the IoT device host.
The gateway device password and the IoT domain device host.
The MQTTX connection described in the common setup.

If you need to create network resources, see Creating a VCN, Creating a Subnet, Creating a NAT Gateway, Creating a VCN Route Table, and Creating an NSG.

Creating the Flow Runtime

In the OCI Console, open IoT, select the IoT domain, select Flow Runtimes, and then select Create Flow Runtime.
Use the following values and wait for the runtime to become active:

Field Value
Display name FR Guide - External MQTT Ingest
Description Subscribes to an external broker, transforms telemetry, and publishes into OCI IoT.
Scale MEDIUM
Subnet <subnet-ocid>
For shared-subscription testing, create a second runtime with the same network configuration. Use display name FR Guide - External MQTT Ingest 2.
Open the runtime editor at https://<flow-runtime-host>.

Field	Value
Display name	`FR Guide - External MQTT Ingest`
Description	Subscribes to an external broker, transforms telemetry, and publishes into OCI IoT.
Scale	`MEDIUM`
Subnet	`<subnet-ocid>`

Importing the Example Flow

Choose either this import path or manual construction. Do not deploy both copies of the scenario: both can ingest the same messages.

Download flows-ingest-external-broker-mqttx.json and save it as a JSON file.
Back up existing flows. In the runtime editor, select Import from the menu, select the downloaded file, and import into a new flow. Do not deploy yet.
The import provides three MQTT-IN nodes, a shared Function node, an MQTT-OUT node, and Debug output. It does not create cloud resources or supply credentials.

Edit the imported broker configurations before deploying:

Value	Configuration
`<external-broker-host>`	In External MQTT Broker, enter the same broker host used by your MQTTX publisher, for example the broker in the common setup. Keep TLS enabled on port 8883, and configure trusted certificates and authentication as required by that broker.
`<unique-client-id>`	Enter a client ID unique to this runtime, such as `fr-guide-fr1`. Do not reuse the publisher's client ID.
`<device-host>`	In IoT Device Host, enter the device host from your IoT domain details. Set the Security username to the gateway external key, `fr-guide-gw-01`, and enter its device password. Retain TLS, port 8883, and MQTT V5.

Retain the MQTT-OUT Content Type of application/json for MQTT V5. Use only synthetic test telemetry with a public broker. TLS does not make a public topic private.

Note

The supplied Function is a demonstration transform: its default expressions replace numeric zero as well as missing values with example values. Use the nonzero test payloads below to validate this example; review the transform's defaulting and validation behavior before using it with operational telemetry.

Review the three input topics: source/gateway/fr-guide-gw-01, source/hvacs/fr-guide-hvac-01, and source/hvacs/fr-guide-hvac-02. Match their external keys to your common scenario setup.
The supplied export uses these exact topics. The manual alternative uses one wildcard input. To use that alternative, replace the three inputs with one input using source/+/+, or the shared subscription described below; do not add a wildcard input alongside overlapping exact-topic inputs.
Deploy only after configuration is complete, then follow Publishing Test Messages. Check MQTT connection status and Debug output. Disable Debug output after validation.

Manually Configuring the Node-RED Flow

Add an mqtt in node and configure it as follows:

Field	Value
Server	`test.mosquitto.org`
Port	`8883`
Use TLS	Enabled
Protocol	MQTT V5
Client ID	`fr-guide-fr1` for the first runtime or `fr-guide-fr2` for the second.
Action	Subscribe to a single topic
Topic for one runtime	`source/+/+`
Topic for two shared runtimes	`$share/fr-guide/source/+/+`
QoS	`1`
Output	Auto-detect a parsed JSON object

Add a TLS configuration. If the runtime trusts the broker's public CA, leave the certificate fields empty. Otherwise, import the required CA certificate.

Add a Function node with one output. Connect the MQTT-IN output to it and use this code:

let topic = msg.topic || "";
let parts = topic.split("/");

// Publishers use source/+/+. This fallback also handles a node
// that exposes the $share prefix in msg.topic.
if (parts[0] === "$share") {
  parts = parts.slice(2);
}

if (parts.length !== 3) {
  node.warn("Ignoring unsupported topic: " + msg.topic);
  return null;
}

const prefix = parts[0];
const deviceType = parts[1];
const externalId = parts[2];

if (prefix !== "source") {
  return null;
}

if (deviceType !== "gateway" && deviceType !== "hvacs") {
  node.warn("Unsupported device type: " + deviceType);
  return null;
}

if (!externalId) {
  return null;
}

if (typeof msg.payload === "string") {
  msg.payload = JSON.parse(msg.payload);
}

msg.payload = msg.payload || {};
msg.payload.time = msg.payload.time || Date.now() * 1000;

if (deviceType === "gateway") {
  msg.topic = "data";
  msg.payload.cpuUtil = Number(msg.payload.cpuUtil || 30);
  msg.payload.memUtil = Number(msg.payload.memUtil || 25);
  msg.payload.diskUtil = Number(msg.payload.diskUtil || 20);
  msg.payload.firmware = msg.payload.firmware || "Oracle Linux 9.1";
} else {
  msg.topic = "hvacs/" + externalId;
  msg.payload.temperature = Number(msg.payload.temperature || 72);
  msg.payload.humidity = Number(msg.payload.humidity || 45);
  msg.payload.pressure = Number(msg.payload.pressure || 101.2);
  msg.payload.mode = msg.payload.mode || "cool";
}

msg.payload = JSON.stringify(msg.payload);
return msg;

Add an mqtt out node, connect the Function output to it, and configure the device-host connection:

Field	Value
Server	`<device-host>`
Port	`8883`
Use TLS	Enabled
Protocol	MQTT V3.1.1 or MQTT V5
Content Type	`application/json` when using MQTT V5.
Topic	Leave blank. The Function node sets `msg.topic`.
QoS	`1`
Username	`fr-guide-gw-01`
Password	Gateway device password

Note

The built-in MQTT-OUT node reads credentials when it creates the broker connection. It doesn't use a password set dynamically on msg.password.

Optionally, connect the Function output to a Debug node.
Select Deploy. Confirm that MQTT-IN connects to the public broker and MQTT-OUT connects to the device host.
To install an exported complete flow document instead, use the following CLI command. This replaces all current flows for the runtime:
```
oci iot flow-runtime update-flows \
  --iot-flow-runtime-id <flow-runtime-ocid> \
  --flows-document file://<path-to-flows-json>
```
See Updating Flows for an IoT Flow Runtime.

Publishing Test Messages

In MQTTX, publish this gateway message to source/gateway/fr-guide-gw-01 with QoS 1:

{
  "cpuUtil": 31,
  "memUtil": 26,
  "diskUtil": 21,
  "firmware": "Oracle Linux 9.1"
}

Publish this message to source/hvacs/fr-guide-hvac-01 with QoS 1:

{
  "temperature": 72.5,
  "humidity": 45.2,
  "pressure": 101.2,
  "mode": "cool"
}

Publish this message to source/hvacs/fr-guide-hvac-02 with QoS 1:

{
  "temperature": 76.1,
  "humidity": 50.0,
  "pressure": 100.8,
  "mode": "heat"
}

Use common Console validation to confirm the gateway and both HVAC devices contain the expected snapshot and normalized values.

Troubleshooting

Verify the broker host, port 8883, TLS trust, subscription topic, QoS, and authentication values.
Use a unique MQTT client ID for each Flow Runtime. Shared runtimes must use the same $share/fr-guide/source/+/+ topic but different client IDs.
Confirm that MQTT-OUT uses the domain device host, the gateway external key as the username, and the gateway device password.
Confirm that the gateway message reaches the device host on data and HVAC messages use hvacs/<external-key>.
Confirm that each HVAC external key exactly matches an active indirectly connected device associated with the gateway.

For more information, see Troubleshooting IoT Flow Runtimes.

FAQs

Does the external broker send data directly to OCI IoT?: No. MQTT-IN subscribes to the external broker, the Function node transforms the topic and message, and MQTT-OUT publishes the result to the IoT domain device host.
Why does the flow rewrite the MQTT topic?: The public topic identifies the source device. The gateway and HVAC adapters expect the IoT device-host topics data and hvacs/<external-key>.
When should I use the shared-subscription topic?: Use $share/fr-guide/source/+/+ when two or more Flow Runtimes should divide messages in one subscription group. Use source/+/+ for one runtime.
Can I use a different MQTT broker?: Yes. Replace the broker connection values and configure the required TLS, authentication, DNS, routes, and security rules for that broker.

Oracle Cloud Infrastructure Documentation