Oracle Cloud Infrastructure Documentation

Scenario: Monitor Normalized IoT Data, Send Commands, and Email Alerts

Dequeue normalized OCI IoT data in Node-RED, send a reset command when Boiler pressure exceeds a threshold, poll the command status, and publish a success, failure, or timeout email alert.

Use this scenario to monitor data already received by OCI IoT and take action when a condition is met.

The flow dequeues normalized data, sends a reset command when Boiler pressure exceeds 100, and polls the command status in the IoT database. A final result continues to the notification nodes. A non-final result loops through a five-second delay and queries the command status again.

In this scenario, IoT data is already inside OCI IoT, so the Flow Runtime acts as a lightweight event processor without requiring an application to access the IoT database directly. You can adapt this pattern for alerting, creating tickets, downstream notifications, and other operational automations.

Complete the common scenario setup before starting.

Prerequisites

  • An OCI Notifications topic and its OCID.
  • A Flow Runtime Resource Principal configuration with the related Notifications and digital twin command-invoke policies.

    Add the Notifications policy to let the Flow Runtime resource principal publish messages to the scenario topic.

    Allow dynamic-group <flow-runtime-dynamic-group> to {ONS_TOPIC_PUBLISH} in compartment <notification-topic-compartment>

    Add the digital twin command-invoke policy to let the resource principal request a command.

    Allow dynamic-group <flow-runtime-dynamic-group> to {IOT_DIGITAL_TWIN_INSTANCE_COMMAND_INVOKE} in compartment <iot-domain-compartment>
  • An IoT domain database connection available to the Dequeue and SQL nodes.
  • The gateway external key and device password recorded when you completed the common digital twin instance setup. The examples use fr-guide-gw-01 as the gateway external key.
  • The OCIDs and external keys of the Boiler digital twin instances monitored by the flow.
  • To validate a successful reset, a device-side command handler that receives the request at boilers/<external-key>/command/reset and returns a response at boilers/<external-key>/command/response.

Step 1: Creating the Monitoring Flow Runtime

  1. In the OCI Console, open the navigation menu, go to Developer Services, under Internet of Things select Domains.
  2. Select the IoT domain you want to work with, select Flow Runtimes, and then select Create Flow Runtime.
  3. Use the following values:
    FieldValue
    Display nameFR Guide - Monitor and Action
    DescriptionDequeues IoT data from NORMALIZED_DATA, sends a Boiler reset command when Boiler pressure exceeds 100, polls for the command result, and publishes an OCI notification.
    ScaleMEDIUM
  4. Select Create and wait a few minutes for the Flow Runtime to become active.

Step 2: Configuring the Monitoring and Command Node-RED Flow

  1. Select the Flow Runtime to open its details page. To open the Node-RED editor, select Open Flow Runtime Editor or select the Editor URL that displays the Flow Runtime endpoint:

    https://<flow-runtime-short-id>.flows.iot.<region>.oci.customer-oci.com/

  2. Add the following nodes to the canvas, and set each node's Name field to the specified value:
    NodeName
    InjectPoll Normalized Data
    DequeueDequeue Normalized Data
    FunctionDetect High Boiler Pressure
    IoT Send CommandSend Boiler Reset Command
    FunctionInitialize Command Status Polling
    DelayWait Before Status Check
    SQLQuery Command Status
    FunctionCheck Command Status
    FunctionFormat Boiler Command Result Notification
    OCI NotificationPublish Boiler Alert Notification
  3. Connect the nodes in this exact order:

    Poll Normalized Data -> Dequeue Normalized Data -> Detect High Boiler Pressure -> Send Boiler Reset Command -> Initialize Command Status Polling -> Wait Before Status Check -> Query Command Status -> Check Command Status

    Connect output 1 of Check Command Status to Format Boiler Command Result Notification, and then connect it to Publish Boiler Alert Notification. Connect output 2 of Check Command Status back to Wait Before Status Check.

    Use one output for the other Function nodes. Output 1 handles a final command result or polling timeout. Output 2 loops a non-final result through the delay and SQL query again.

  4. Configure the Poll Normalized Data Inject node with these settings:
    FieldValue
    NamePoll Normalized Data
    PayloadTimestamp
    TopicLeave blank.
    Inject once afterSelect the checkbox and use 0.1 seconds.
    RepeatInterval
    Every10 seconds

    Use the 10-second interval for this walkthrough. After validating the flow, adjust the interval to match the expected queue traffic and database load.

  5. Configure the Dequeue Normalized Data Dequeue node:
    FieldValue
    NameDequeue Normalized Data
    DB ConnectionThe IoT domain database connection.
    ModeTransactional
    Queue name<domain-short-id>__IOT.NORMALIZED_DATA
    Subscriber<domain-short-id>_<flow-runtime-short-id>_SUBSCRIBER
    Payload typeJSON
    Dequeue modeRemove
    Wait0
    Batch size2
  6. Set the Function node Name to Detect High Boiler Pressure and Outputs to 1. Replace the example digital twin OCIDs in boilerExternalKeyMap. NORMALIZED_DATA can contain pressure events from both HVAC and Boiler devices. Use this code to send a reset command only when the event belongs to a mapped Boiler digital twin instance and its pressure is greater than 100: 
    const PRESSURE_LIMIT = 100;

// Include only Boiler digital twins in this map.
// HVAC pressure records are ignored because their OCIDs are not mapped here.
const boilerExternalKeyMap = {
  "<boiler-01-digital-twin-ocid>": "fr-guide-boiler-01",
  "<boiler-02-digital-twin-ocid>": "fr-guide-boiler-02"
};

const event = msg.payload || {};
const twinOcid = event.digitalTwinInstanceId;
const externalKey = boilerExternalKeyMap[twinOcid];

if (event.contentPath !== "pressure" || !externalKey) {
  return null;
}

const pressure = Number(event.value);
if (!Number.isFinite(pressure) || pressure <= PRESSURE_LIMIT) {
  return null;
}

msg.digitalTwinOcid = twinOcid;
msg.requestEndpoint = `boilers/${externalKey}/command/reset`;
msg.responseEndpoint = `boilers/${externalKey}/command/response`;

msg.title = `Boiler High Pressure Alert - ${externalKey}`;
msg.notificationpayload = {
  alert: "Boiler pressure is above the allowed limit",
  action: "Reset boiler command sent",
  digitalTwinOcid: twinOcid,
  externalKey: externalKey,
  pressure: pressure,
  threshold: PRESSURE_LIMIT,
  timeObserved: event.timeObserved
};

msg.payload = { reset: true };
return msg;
  7. Configure the Send Boiler Reset Command IoT Send Command node:
    FieldValue
    NameSend Boiler Reset Command
    OCI ConfigFR - Resource Principal
    Digital twin OCIDLeave blank. The Detect High Boiler Pressure node sets msg.digitalTwinOcid.
    Request endpointLeave blank. The Detect High Boiler Pressure node sets msg.requestEndpoint.
    Request payloadLeave blank. The Detect High Boiler Pressure node sets msg.payload to {"reset":true}.
    Wait for responseSelect the checkbox.
    Response endpointLeave blank. The Detect High Boiler Pressure node sets msg.responseEndpoint.
    Request durationPT1M
    Response durationPT1M

    When the node sends the command, it adds the command record ID to msg.rawCommandDataRecordId.

  8. Set the Function node Name to Initialize Command Status Polling and Outputs to 1. Use this code to start a 70-second polling deadline and initialize the poll counter: 
    msg.pollDeadline = Date.now() + 70000;
msg.pollCount = 0;
return msg;

    The 70-second deadline covers the one-minute response window plus ten seconds for the final database update.

  9. Configure the Wait Before Status Check Delay node to delay each message for 5 seconds.
    FieldValue
    NameWait Before Status Check
    ActionDelay each message, fixed delay
    For5 seconds

    This node receives the initial polling message from Initialize Command Status Polling and non-final results from output 2 of Check Command Status.

  10. Configure the Query Command Status SQL node to query the command response:
    FieldValue
    NameQuery Command Status
    DB ConnectionThe IoT domain database connection.
    SQL sourceEditor
    Binds sourceEditor
    Bind variablerecordId
    Bind sourceMessage property
    Property pathrawCommandDataRecordId
    Max rows1

    Replace <domain-short-id> with the domain short ID from the device host, and use this query:

    SELECT
       ID,
       DIGITAL_TWIN_INSTANCE_ID,
       REQUEST_ENDPOINT,
       RESPONSE_ENDPOINT,
       DELIVERY_STATUS,
       UPPER(
           JSON_VALUE(
               RESPONSE_DATA FORMAT JSON,
               '$.reset'
               RETURNING VARCHAR2(30)
               NULL ON ERROR
           )
       ) AS RESET_RESULT,
       TIME_CREATED,
       TIME_UPDATED,
       TIME_FINISHED
  FROM <domain-short-id>__IOT.RAW_COMMAND_DATA
 WHERE ID = :recordId

    For column descriptions, see RAW_COMMAND_DATA.

  11. Set the Function node Name to Check Command Status and Outputs to 2. Connect output 1 to Format Boiler Command Result Notification and output 2 back to Wait Before Status Check. Use this code to route a final result or timeout through output 1 and a non-final result through output 2: 
    const rows = Array.isArray(msg.payload) ? msg.payload : [];
const row = rows[0];
msg.pollCount = (msg.pollCount || 0) + 1;

if (!row) {
  if (Date.now() < msg.pollDeadline) {
    return [null, msg];
  }

  msg.payload = [{
    DELIVERY_STATUS: "POLL_TIMEOUT",
    RESET_RESULT: null,
    TIME_FINISHED: null
  }];
  return [msg, null];
}

const status = String(row.DELIVERY_STATUS || "").toUpperCase();
const finalStatuses = [
  "COMPLETED",
  "REJECTED",
  "REFUSED",
  "EXPIRED",
  "BAD_RESPONSE",
  "NOT_RESPONDED"
];

if (finalStatuses.includes(status)) {
  return [msg, null];
}

if (Date.now() >= msg.pollDeadline) {
  msg.payload = [{
    ...row,
    DELIVERY_STATUS: "POLL_TIMEOUT"
  }];
  return [msg, null];
}

return [null, msg];
  12. Set the Function node Name to Format Boiler Command Result Notification and Outputs to 1. Use this code to create a success, failure, or timeout email body: 
    const row = msg.payload[0] || {};
const alert = msg.notificationpayload || {};
const commandStatus = String(row.DELIVERY_STATUS || "UNKNOWN").toUpperCase();
const resetResult = String(row.RESET_RESULT || "NO_RESPONSE").toUpperCase();
const resetSucceeded = commandStatus === "COMPLETED" && resetResult === "SUCCESS";

let result;
if (resetSucceeded) {
  msg.title = `Boiler High Pressure Alert - Reset Successful - ${alert.externalKey}`;
  result = "The device confirmed that the boiler reset completed successfully.";
} else if (commandStatus === "POLL_TIMEOUT") {
  msg.title = `Boiler High Pressure Alert - Reset Status Timed Out - ${alert.externalKey}`;
  result = "The boiler did not respond to the reset command within the allowed time.";
} else {
  msg.title = `Boiler High Pressure Alert - Reset Failed - ${alert.externalKey}`;
  result = "The device did not confirm a successful boiler reset.";
}

msg.payload = JSON.stringify({
  ...alert,
  result: result,
  commandStatus: commandStatus,
  resetResult: resetResult,
  timeFinished: row.TIME_FINISHED || null
}, null, 2);

return msg;
  13. Configure the Publish Boiler Alert Notification OCI Notification node:
    FieldValue
    NamePublish Boiler Alert Notification
    OCI ConfigFR - Resource Principal
    Topic OCID<notification-topic-ocid>
    TitleLeave blank. The Format Boiler Command Result Notification node sets msg.title.
    BodyLeave blank. The Format Boiler Command Result Notification node sets msg.payload.
  14. Optionally connect Debug nodes while validating the dequeued record, command record ID, SQL result, and notification payload.
  15. Select Deploy.
  16. To install an exported complete flow document instead, use: 
    oci iot flow-runtime update-flows \
  --iot-flow-runtime-id <flow-runtime-ocid> \
  --flows-document file://<path-to-flows-json>

    For more information on using the Console or the API to update the flow document, see Updating Flows for an IoT Flow Runtime.

Step 3: Creating an Email Subscription

  1. In the OCI Console, open the navigation menu and go to Developer Services. Under Application Integration, select Notifications, and then select the notification topic configured in the flow.
  2. Select Subscriptions, and then select Create Subscription.
  3. Select the Email protocol, enter the email address that receives Boiler alerts, and create the subscription.
  4. Open the confirmation email and follow the confirmation link to confirm the subscription.
  5. In Notifications, verify that the subscription state is Active, not Pending.

    For more information, including how to use the CLI and API, see Creating a Topic in Notifications and Creating an Email Subscription.

Step 4: Subscribing to the Command Request Endpoint

  1. In MQTTX or another MQTT client, connect to the OCI IoT device host using the gateway credentials:
    FieldValue
    Host<device-host>
    Port8883
    SSL/TLSEnabled
    ProtocolMQTT V3.1.1 or MQTT V5
    Usernamefr-guide-gw-01
    PasswordThe gateway device password.
  2. Subscribe to the command request endpoint using these settings:
    FieldValue
    Topicboilers/fr-guide-boiler-01/command/reset
    QoS1

    Keep the Gateway MQTT client connected and actively subscribed before triggering the flow. If no connected client is subscribed to the exact request endpoint, the raw command finishes immediately with DELIVERY_STATUS set to REFUSED.

Step 5: Generating a High Boiler Pressure Message

Use either HTTPS or MQTTS to send the high-pressure Boiler message. Keep the Gateway MQTT client from the previous step connected and subscribed so that it can receive the reset command.

  • Use the gateway external key as the device user name and the gateway device password as the password. Replace <domain-short-id-from-device-host>, <region>, and <gateway-device-password> with the values for your environment.

    curl -i -X POST \
  -u "fr-guide-gw-01:<gateway-device-password>" \
  -H "Content-Type: application/json" \
  "https://<domain-short-id-from-device-host>.device.iot.<region>.oci.oraclecloud.com/boilers/fr-guide-boiler-01" \
  -d '{
    "temperature": 83,
    "pressure": 102
  }'

  • Using the same Gateway-authenticated MQTT connection, publish a high-pressure Boiler message with these settings:

    FieldValue
    Topicboilers/fr-guide-boiler-01
    QoS1
    Payload typeJSON
    {
  "temperature": 83,
  "pressure": 102
}

Step 6: Sending the Command Response

  1. Wait for the MQTT client to receive the reset request on boilers/fr-guide-boiler-01/command/reset, and confirm that the request payload contains {"reset":true}.
  2. Publish the successful device response using these settings:
    FieldValue
    Topicboilers/fr-guide-boiler-01/command/response
    QoS1
    Payload typeJSON
    {
  "reset": "SUCCESS"
}

Step 7: Validating the Polling Flow and Email Notification

  1. Confirm that OCI IoT ingests the high-pressure Boiler message sent over HTTPS or MQTTS and that the monitoring flow dequeues the individual events from NORMALIZED_DATA.
  2. Confirm that Detect High Boiler Pressure ignores HVAC pressure records and triggers only for a mapped Boiler digital twin instance whose pressure is greater than 100.
  3. Confirm that Send Boiler Reset Command sends {"reset":true} to boilers/fr-guide-boiler-01/command/reset and returns msg.rawCommandDataRecordId.
  4. Confirm that Initialize Command Status Polling sets the 70-second deadline and that Query Command Status uses the command record ID to query RAW_COMMAND_DATA every five seconds.
  5. For the successful response, confirm that the query returns DELIVERY_STATUS as COMPLETED and RESET_RESULT as SUCCESS. Confirm that output 1 of Check Command Status exits the polling loop immediately.
  6. Confirm that the active email subscription receives an email with subject Boiler High Pressure Alert - Reset Successful - fr-guide-boiler-01.
  7. Confirm that the email body resembles: 
    {
  "alert": "Boiler pressure is above the allowed limit",
  "action": "Reset boiler command sent",
  "digitalTwinOcid": "<boiler-digital-twin-ocid>",
  "externalKey": "fr-guide-boiler-01",
  "pressure": 102,
  "threshold": 100,
  "timeObserved": "<timestamp>",
  "result": "The device confirmed that the boiler reset completed successfully.",
  "commandStatus": "COMPLETED",
  "resetResult": "SUCCESS",
  "timeFinished": "<timestamp>"
}
  8. For a final status that does not confirm a successful reset, confirm that output 1 exits the loop and the email subject reports Boiler High Pressure Alert - Reset Failed. If no final status is available within 70 seconds, confirm that the flow emits POLL_TIMEOUT and the email subject reports Boiler High Pressure Alert - Reset Status Timed Out.
  9. On the Boiler digital twin instance Data tab, confirm that the latest snapshot shows pressure with a value of 102 and the expected observation time.
    For more information, see Getting Digital Twin Instance Content.

Security Considerations

Use the Flow Runtime Resource Principal to publish only to approved Notifications topics and invoke commands only on approved digital twin instances. Limit the notification and command payloads to the operational fields that recipients require, and protect the IoT domain database connection, queue subscriber, and command response data.

Troubleshooting

  • Confirm the queue name, subscriber, JSON payload type, wait value, batch size of 2, and IoT domain database connection.
  • Use a Debug node to confirm that the normalized record exposes digitalTwinInstanceId, contentPath, value, and timeObserved.
  • Confirm that boilerExternalKeyMap contains only the exact Boiler digital twin OCIDs and matching digital twin instance external keys. HVAC pressure records are ignored because their OCIDs aren't mapped. To find an external key, see Getting a Digital Twin's Instance Details or to change the external key, see Updating a Digital Twin Instance.
  • Confirm that the IoT Send Command node uses the Resource Principal configuration, receives msg.digitalTwinOcid, msg.requestEndpoint, and msg.responseEndpoint, and has Wait for response selected.
  • Confirm that the Gateway MQTT client stays connected and subscribed to the exact request endpoint before the flow sends the command. Without an active subscriber, the command immediately finishes with DELIVERY_STATUS set to REFUSED.
  • Confirm that the device-side command handler listens on the request endpoint and publishes its result to the response endpoint before the response duration expires.
  • Confirm that Initialize Command Status Polling sets msg.pollDeadline and msg.pollCount before the first delay.
  • If the SQL node returns no row, confirm that msg.rawCommandDataRecordId is present after the IoT Send Command node and is mapped to the recordId bind variable. Before the deadline, a missing row must continue through output 2 of Check Command Status.
  • Confirm that output 1 of Check Command Status connects to Format Boiler Command Result Notification and output 2 connects back to Wait Before Status Check.
  • If the reset is reported as failed, inspect DELIVERY_STATUS, RESET_RESULT, and TIME_FINISHED in RAW_COMMAND_DATA. The final statuses are COMPLETED, REJECTED, REFUSED, EXPIRED, BAD_RESPONSE, and NOT_RESPONDED.
  • If the flow reports POLL_TIMEOUT, confirm that the Delay node uses five seconds and that the command did not reach a final status before the 70-second deadline.
  • Confirm the Notifications topic OCID, Resource Principal configuration, permission to publish messages, and active email subscription.

For more information, see Troubleshooting IoT Flow Runtimes and Flow Runtimes FAQs.

FAQs

Why must the email subscription be active?
OCI Notifications doesn't deliver the alert to an email subscription while its state is Pending. Follow the confirmation link before testing the flow.
Why does the flow query the command status every five seconds?
The short delay lets the flow detect a final command status without waiting for the complete response window. A non-final result loops back to the Delay node and is queried again.
Why is the polling deadline 70 seconds?
The command response duration is PT1M. The additional ten seconds allow OCI IoT to apply the final database update before the flow reports POLL_TIMEOUT.
Which queue records does this flow evaluate?
The flow dequeues JSON records from <domain-short-id>__IOT.NORMALIZED_DATA. It evaluates pressure events only when the digital twin OCID is present in boilerExternalKeyMap, so HVAC pressure records and unmapped Boiler records are ignored.
When does the notification report a successful reset?
The reset succeeds only when DELIVERY_STATUS is COMPLETED and the response JSON contains reset: SUCCESS. Other final results use the failed-reset notification, while a missing final result at the deadline uses the timeout notification.
Why did the command finish with REFUSED?
The Gateway MQTT client wasn't connected and actively subscribed to the exact command request endpoint when the flow sent the command. Subscribe to boilers/fr-guide-boiler-01/command/reset before triggering the flow.
Can I generate the test data through the live-ingestion scenario?
Yes. This scenario publishes directly to the IoT device-host topic boilers/fr-guide-boiler-01. If you keep the live broker ingestion flow running instead, publish the same payload to source/boilers/fr-guide-boiler-01 on the public broker.