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.

Tasks

Create the Monitoring OCI Flow Runtime
Configure the Monitoring Node-RED Flow
Create the Email Subscription
Subscribe to the Command Request Endpoint
Generate a High Boiler Pressure Message
Send the Command Response
Validate the Polling Flow and Email Notification

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

In the OCI Console, open the navigation menu, go to Developer Services, under Internet of Things select Domains.
Select the IoT domain you want to work with, select Flow Runtimes, and then select Create Flow Runtime.

Use the following values:

Field	Value
Display name	`FR Guide - Monitor and Action`
Description	Dequeues 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.
Scale	`MEDIUM`

Select Create and wait a few minutes for the Flow Runtime to become active.

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

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/

Add the following nodes to the canvas, and set each node's Name field to the specified value:

Node	Name
Inject	`Poll Normalized Data`
Dequeue	`Dequeue Normalized Data`
Function	`Detect High Boiler Pressure`
IoT Send Command	`Send Boiler Reset Command`
Function	`Initialize Command Status Polling`
Delay	`Wait Before Status Check`
SQL	`Query Command Status`
Function	`Check Command Status`
Function	`Format Boiler Command Result Notification`
OCI Notification	`Publish Boiler Alert Notification`

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.
Configure the Poll Normalized Data Inject node with these settings:

Field Value
Name Poll Normalized Data
Payload Timestamp
Topic Leave blank.
Inject once after Select the checkbox and use 0.1 seconds.
Repeat Interval
Every 10 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.

Field	Value
Name	`Poll Normalized Data`
Payload	Timestamp
Topic	Leave blank.
Inject once after	Select the checkbox and use `0.1 seconds`.
Repeat	Interval
Every	`10 seconds`

Configure the Dequeue Normalized Data Dequeue node:

Field	Value
Name	`Dequeue Normalized Data`
DB Connection	The IoT domain database connection.
Mode	Transactional
Queue name	`<domain-short-id>__IOT.NORMALIZED_DATA`
Subscriber	`<domain-short-id>_<flow-runtime-short-id>_SUBSCRIBER`
Payload type	JSON
Dequeue mode	Remove
Wait	`0`
Batch size	`2`

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;

Configure the Send Boiler Reset Command IoT Send Command node:

Field	Value
Name	`Send Boiler Reset Command`
OCI Config	`FR - Resource Principal`
Digital twin OCID	Leave blank. The Detect High Boiler Pressure node sets `msg.digitalTwinOcid`.
Request endpoint	Leave blank. The Detect High Boiler Pressure node sets `msg.requestEndpoint`.
Request payload	Leave blank. The Detect High Boiler Pressure node sets `msg.payload` to `{"reset":true}`.
Wait for response	Select the checkbox.
Response endpoint	Leave blank. The Detect High Boiler Pressure node sets `msg.responseEndpoint`.
Request duration	`PT1M`
Response duration	`PT1M`

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

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.
Configure the Wait Before Status Check Delay node to delay each message for 5 seconds.

Field Value
Name Wait Before Status Check
Action Delay each message, fixed delay
For 5 seconds

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

Field	Value
Name	`Wait Before Status Check`
Action	Delay each message, fixed delay
For	`5 seconds`

Configure the Query Command Status SQL node to query the command response:

Field	Value
Name	`Query Command Status`
DB Connection	The IoT domain database connection.
SQL source	Editor
Binds source	Editor
Bind variable	`recordId`
Bind source	Message property
Property path	`rawCommandDataRecordId`
Max rows	`1`

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.

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];

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;

Configure the Publish Boiler Alert Notification OCI Notification node:

Field	Value
Name	`Publish Boiler Alert Notification`
OCI Config	`FR - Resource Principal`
Topic OCID	`<notification-topic-ocid>`
Title	Leave blank. The Format Boiler Command Result Notification node sets `msg.title`.
Body	Leave blank. The Format Boiler Command Result Notification node sets `msg.payload`.

Optionally connect Debug nodes while validating the dequeued record, command record ID, SQL result, and notification payload.
Select Deploy.
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 4: Subscribing to the Command Request Endpoint

In MQTTX or another MQTT client, connect to the OCI IoT device host using the gateway credentials:

Field Value
Host <device-host>
Port 8883
SSL/TLS Enabled
Protocol MQTT V3.1.1 or MQTT V5
Username fr-guide-gw-01
Password The gateway device password.
Subscribe to the command request endpoint using these settings:

Field Value
Topic boilers/fr-guide-boiler-01/command/reset
QoS 1

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.

Field	Value
Host	`<device-host>`
Port	`8883`
SSL/TLS	Enabled
Protocol	MQTT V3.1.1 or MQTT V5
Username	`fr-guide-gw-01`
Password	The gateway device password.

Field	Value
Topic	`boilers/fr-guide-boiler-01/command/reset`
QoS	`1`

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:

Field Value
Topic boilers/fr-guide-boiler-01
QoS 1
Payload type JSON
```
{
  "temperature": 83,
  "pressure": 102
}
```

Field	Value
Topic	`boilers/fr-guide-boiler-01`
QoS	`1`
Payload type	JSON

Step 6: Sending the Command Response

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}.
Publish the successful device response using these settings:

Field Value
Topic boilers/fr-guide-boiler-01/command/response
QoS 1
Payload type JSON
```
{
  "reset": "SUCCESS"
}
```

Field	Value
Topic	`boilers/fr-guide-boiler-01/command/response`
QoS	`1`
Payload type	JSON

Step 7: Validating the Polling Flow and Email Notification

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.
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.
Confirm that Send Boiler Reset Command sends {"reset":true} to boilers/fr-guide-boiler-01/command/reset and returns msg.rawCommandDataRecordId.
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.
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.
Confirm that the active email subscription receives an email with subject Boiler High Pressure Alert - Reset Successful - fr-guide-boiler-01.

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>"
}

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

Oracle Cloud Infrastructure Documentation