Oracle Cloud Infrastructure Documentation

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.

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

  1. In the OCI Console, open IoT, select the IoT domain, select Flow Runtimes, and then select Create Flow Runtime.
  2. Use the following values and wait for the runtime to become active:
    FieldValue
    Display nameFR Guide - External MQTT Ingest
    DescriptionSubscribes to an external broker, transforms telemetry, and publishes into OCI IoT.
    ScaleMEDIUM
    Subnet<subnet-ocid>
  3. For shared-subscription testing, create a second runtime with the same network configuration. Use display name FR Guide - External MQTT Ingest 2.
  4. Open the runtime editor at https://<flow-runtime-host>.

Configuring the Node-RED Flow

  1. Add an mqtt in node and configure it as follows:
    FieldValue
    Servertest.mosquitto.org
    Port8883
    Use TLSEnabled
    ProtocolMQTT V5
    Client IDfr-guide-fr1 for the first runtime or fr-guide-fr2 for the second.
    ActionSubscribe to a single topic
    Topic for one runtimesource/+/+
    Topic for two shared runtimes$share/fr-guide/source/+/+
    QoS1
    OutputAuto-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.

  2. 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;
  3. Add an mqtt out node, connect the Function output to it, and configure the device-host connection:
    FieldValue
    Server<device-host>
    Port8883
    Use TLSEnabled
    ProtocolMQTT V3.1.1 or MQTT V5
    TopicLeave blank. The Function node sets msg.topic.
    QoS1
    Usernamefr-guide-gw-01
    PasswordGateway 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.
  4. Optionally, connect the Function output to a Debug node.
  5. Select Deploy. Confirm that MQTT-IN connects to the public broker and MQTT-OUT connects to the device host.
  6. 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

  1. 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"
}
  2. Publish this message to source/hvacs/fr-guide-hvac-01 with QoS 1:
    {
  "temperature": 72.5,
  "humidity": 45.2,
  "pressure": 101.2,
  "mode": "cool"
}
  3. Publish this message to source/hvacs/fr-guide-hvac-02 with QoS 1:
    {
  "temperature": 76.1,
  "humidity": 50.0,
  "pressure": 100.8,
  "mode": "heat"
}
  4. 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.