シナリオ: 外部ブローカからのライブ・データの取込み

外部MQTTブローカをサブスクライブし、Node-REDでゲートウェイおよびHVACメッセージを変換して、OCI IoTに取り込みます。

このシナリオは、デバイスまたはゲートウェイが外部MQTTブローカにテレメトリをすでに公開している場合に使用します。フローはMQTT-IN -> Function -> MQTT-OUTで、オプションのDebugブランチがあります。

タスク

フロー・ランタイムの作成
ノード-REDフローの構成
テスト・メッセージの公開

前提条件

開始する前に、一般的なシナリオ設定を完了してください。この例では、test.mosquitto.org:8883からデータを受信し、パブリック・トピックをアダプタ・トピックに変換して、OCI IoTデバイス・ホストに公開します。
0.0.0.0/0のNATゲートウェイを介したルート、またはランタイムが外部ブローカに到達できるようにする別の承認済ルートを含むプライベート・サブネット。
TCPポート8883およびIoTデバイス・ホスト上のブローカへのアウトバウンドTLS接続を許可するネットワーク・セキュリティ・ルール。
ゲートウェイ・デバイス・パスワードおよびIoTドメイン・デバイス・ホスト。
共通設定で説明されているMQTTX接続。

ネットワーク・リソースを作成する必要がある場合は、VCNの作成、サブネットの作成、NAT Gatewayの作成、VCNルート表の作成およびNSGの作成を参照してください。

フロー・ランタイムの作成

OCIコンソールで、IoTを開き、IoTドメインを選択し、「フロー・ランタイム」、「フロー・ランタイムの作成」の順に選択します。

次の値を使用し、ランタイムがアクティブになるまで待機します。

フィールド	値
表示名	`FR Guide - External MQTT Ingest`
摘要	外部ブローカーを購読し、テレメトリを変換し、OCI IoTに公開します。
スケール	`MEDIUM`
サブネット	`<subnet-ocid>`

共有サブスクリプション・テストの場合は、同じネットワーク構成で2番目のランタイムを作成します。表示名FR Guide - External MQTT Ingest 2を使用します。
https://<flow-runtime-host>でランタイム・エディタを開きます。

ノード-REDフローの構成

ノードに mqttを追加し、次のように構成します。

フィールド	値
サーバー	`test.mosquitto.org`
ポート	`8883`
TLSの使用	有効
プロトコル	MQTT V5
クライアントID	最初のランタイムの場合は`fr-guide-fr1`、2番目のランタイムの場合は`fr-guide-fr2`。
処理	単一のトピックへのサブスクライブ
1つのランタイムのトピック	`source/+/+`
2つの共有ランタイムのトピック	`$share/fr-guide/source/+/+`
QoS	`1`
出力	解析されたJSONオブジェクトの自動検出

TLS構成を追加します。ランタイムがブローカのパブリックCAを信頼する場合は、証明書フィールドを空のままにします。それ以外の場合は、必要なCA証明書をインポートします。

1つの出力を含む関数ノードを追加します。MQTT-IN出力を接続し、次のコードを使用します。

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;

mqtt outノードを追加し、Function出力をそれに接続して、デバイスホスト接続を構成します。

フィールド	値
サーバー	`<device-host>`
ポート	`8883`
TLSの使用	有効
プロトコル	MQTT V3.1.1またはMQTT V5
トピック	空白のまま。「ファンクション」ノードは`msg.topic`を設定します。
QoS	`1`
ユーザ名	`fr-guide-gw-01`
パスワード	ゲートウェイ・デバイスのパスワード

ノート

組込みMQTT-OUTノードは、ブローカ接続の作成時に資格証明を読み取ります。msg.passwordで動的に設定されたパスワードは使用しません。

オプションで、関数出力をデバッグ・ノードに接続します。
「デプロイ」を選択しますMQTT-INがパブリック・ブローカに接続され、MQTT-OUTがデバイス・ホストに接続されていることを確認します。
かわりにエクスポートされた完全なフロー・ドキュメントをインストールするには、次のCLIコマンドを使用します。これにより、ランタイムの現在のフローがすべて置き換えられます。
```
oci iot flow-runtime update-flows \
  --iot-flow-runtime-id <flow-runtime-ocid> \
  --flows-document file://<path-to-flows-json>
```
IoTフロー・ランタイムのフローの更新を参照してください。

テスト・メッセージの公開

MQTTXで、QoS 1を使用してこのゲートウェイ・メッセージをsource/gateway/fr-guide-gw-01に公開します。
```
{
  "cpuUtil": 31,
  "memUtil": 26,
  "diskUtil": 21,
  "firmware": "Oracle Linux 9.1"
}
```
QoS 1を使用して、次のメッセージをsource/hvacs/fr-guide-hvac-01に公開します。
```
{
  "temperature": 72.5,
  "humidity": 45.2,
  "pressure": 101.2,
  "mode": "cool"
}
```
QoS 1を使用して、次のメッセージをsource/hvacs/fr-guide-hvac-02に公開します。
```
{
  "temperature": 76.1,
  "humidity": 50.0,
  "pressure": 100.8,
  "mode": "heat"
}
```
共通コンソール検証を使用してゲートウェイを確認し、両方のHVACデバイスに予想されるスナップショットと正規化された値が含まれていることを確認します。

トラブルシューティング

ブローカ・ホスト、ポート8883、TLS信頼、サブスクリプション・トピック、QoSおよび認証の値を確認します。
各フロー・ランタイムに一意のMQTTクライアントIDを使用します。共有ランタイムでは、同じ$share/fr-guide/source/+/+トピックを使用する必要がありますが、クライアントIDは異なります。
MQTT-OUTでドメイン・デバイス・ホスト、ユーザー名としてゲートウェイ外部キー、およびゲートウェイ・デバイス・パスワードが使用されていることを確認します。
ゲートウェイ・メッセージがdata上のデバイス・ホストに到達し、HVACメッセージでhvacs/<external-key>が使用されていることを確認します。
各HVAC外部キーが、ゲートウェイに関連付けられたアクティブな間接接続デバイスと完全に一致することを確認します。

詳細は、「IoTフロー・ランタイムのトラブルシューティング」を参照してください。

FAQ

外部ブローカはOCI IoTにデータを直接送信しますか?: 番号MQTT-INは外部ブローカをサブスクライブし、ファンクション・ノードはトピックとメッセージを変換し、MQTT-OUTは結果をIoTドメイン・デバイス・ホストにパブリッシュします。
フローによってMQTTトピックがリライトされるのはなぜですか。: パブリック・トピックはソース・デバイスを識別します。ゲートウェイおよびHVACアダプタは、IoTデバイス・ホスト・トピックdataおよびhvacs/<external-key>を想定しています。
共有サブスクリプション・トピックを使用する必要があるのはどのような場合ですか。: 2つ以上のフロー・ランタイムが1つのサブスクリプション・グループでメッセージを分割する必要がある場合、$share/fr-guide/source/+/+を使用します。1つのランタイムにsource/+/+を使用します。
別のMQTTブローカを使用できますか?: はい。ブローカ接続値を置換し、そのブローカに必要なTLS、認証、DNS、ルートおよびセキュリティ・ルールを構成します。

Oracle Cloud Infrastructureドキュメント