# EMQXでNode-REDを使う

[Node-RED](https://nodered.org/)は、ハードウェアデバイス、API、オンラインサービスをブラウザベースのエディターで接続するフローベースのプログラミングツールです。視覚的なノードベースのインターフェースを使い、あらかじめ用意されたノードをつなげてデータフローを作成します。Node-REDは組み込みの`mqtt-in`（サブスクライブ）および`mqtt-out`（パブリッシュ）ノードを通じてMQTTをネイティブにサポートしており、EMQXからのIoTデータ処理に広く利用されています。

このページでは、Node-REDのインストール方法、EMQXへの接続方法、MQTTメッセージを解析・フィルタリング・変換するデータ処理パイプラインの構築方法を説明します。

## 前提条件

- Node.js 18 LTS または 20 LTS（NPMインストール用）
- EMQXのデプロイ環境、またはテスト用にEMQXパブリックブローカーを利用
- テストメッセージ送信用の[MQTTX](https://mqttx.app/)などのMQTTクライアント

## Node-REDのインストール

**NPM経由の場合:**

```bash
npm install -g --unsafe-perm node-red
```

続いてNode-REDを起動します:

```bash
node-red
```

**Docker経由の場合:**

```bash
docker run -it -p 1880:1880 --name mynodered nodered/node-red
```

起動後、ブラウザで`http://127.0.0.1:1880`にアクセスするとNode-REDエディターが開きます。

![Node-REDエディターへのアクセス](./assets/access_node_red_editor.png)

> Raspberry Piやクラウドデプロイなど、その他のインストール方法については[Node-REDドキュメント](https://nodered.org/docs/getting-started/)をご覧ください。

## MQTTブローカーの設定

Node-REDを接続するためにMQTTブローカーが必要です。本ガイドではMQTT 3.1、3.1.1、5.0をサポートするEMQXを使用します。

### EMQXパブリックブローカー（テスト用）

独自のブローカーをデプロイせずに手軽にテストしたい場合は、EMQXパブリックブローカーを利用できます。

| パラメーター       | 値                |
| ------------------ | ----------------- |
| ブローカーアドレス | `broker.emqx.io`  |
| TCPポート          | `1883`            |
| SSL/TLSポート      | `8883`            |
| WebSocketポート    | `8083`            |
| セキュアWebSocketポート | `8084`        |

パブリックブローカーはテストおよびデモ目的のみでの利用を想定しています。

### EMQX Enterpriseデプロイメント

本番環境では、自身のEMQX EnterpriseデプロイメントにNode-REDを接続します。ブローカーのアドレス、ポート、認証情報は環境に応じて設定してください。

一般的な構成例：

- カスタムのブローカーホスト名またはIPアドレス
- ユーザー名/パスワード認証または相互TLS認証
- トピックに適用されるアクセス制御ルール（ACL）

Node-REDのブローカー接続設定時には、EMQX Enterpriseのリスナーおよび認証設定を参照してください。

> 自己管理型EMQX Enterpriseのほか、完全マネージドMQTTサービスの[EMQX Cloud](https://docs.emqx.com/en/cloud/latest/)（ServerlessまたはDedicated）にもNode-REDを接続可能です。EMQX Cloudから提供されるブローカーアドレス、ポート、認証情報を使用してください。

## 基本的なMQTTフローの構築

以下の手順で、1つのトピックをサブスクライブし、受信したメッセージを別のトピックに転送する最小限のフローを作成します。

### ステップ1: MQTTサブスクライブノードの追加

1. Node-REDエディターで、左側パレットから**mqtt-in**ノードをキャンバスにドラッグします。

2. ノードをダブルクリックしてプロパティを開きます。

3. **Server**フィールド横の鉛筆アイコンをクリックし、新しいブローカー接続を作成します。

4. **Server**アドレスに`broker.emqx.io`を入力し、**Add**をクリックします。

   ![サブスクライブノードの設定](./assets/config_subscribe_node.png)

5. **Topic**に`test/node_red/in`を設定します。

6. 必要に応じて**QoS**レベルを設定し、**Done**をクリックします。

   ![トピックのサブスクライブ設定](./assets/subscribe_to_topic.png)

### ステップ2: MQTTパブリッシュノードの追加

1. **mqtt-out**ノードをキャンバスにドラッグします。

2. ノードをダブルクリックしてプロパティを開きます。

3. ステップ1で設定したブローカーを**Server**ドロップダウンから選択します。

4. **Topic**に`test/node_red/out`を設定します。

5. 必要に応じて**QoS**と**Retain**を設定し、**Done**をクリックします。

   ![パブリッシュノードの設定](./assets/config_publish_node.png)

### ステップ3: 接続とデプロイ

1. **mqtt-in**ノードの出力ポートから**mqtt-out**ノードの入力ポートへワイヤーを引きます。

2. 右上の**Deploy**ボタンをクリックします。

3. 両ノードが緑色の**connected**ステータスを表示していることを確認します。

これで、`test/node_red/in`で受信したすべてのメッセージが`test/node_red/out`に転送されるフローが完成しました。

![ノードの接続](./assets/connect_nodes.png)

## 高度なデータ処理パイプラインの構築

Node-REDの真価は複数ノードを連結してデータをフィルタリングや変換し、再パブリッシュできる点にあります。以下の例では、

1. MQTT経由でJSON形式のセンサーデータを受信
2. 生ペイロードをJavaScriptオブジェクトにパース
3. 重複する温度データをフィルタリング
4. 結果を整形して再パブリッシュ

というパイプラインを構築します。

フロー全体は以下の通りです：**mqtt-in** -> **json** -> **rbe** -> **template** -> **mqtt-out**

### ステップ1: JSONノードの追加

1. パレットから**json**ノードをキャンバスにドラッグします。

2. ダブルクリックして設定を開き、**Action**を**Always Convert to JavaScript Object**に設定します。

3. **Done**をクリックします。

4. **mqtt-in**ノードの出力を**json**ノードの入力に接続します。

これにより、受信したペイロードがJavaScriptオブジェクトにパースされ、下流のノードで`msg.payload.temperature`などの個別フィールドにアクセス可能になります。

![jsonノードの接続](./assets/connect_json_nodes.png)

### ステップ2: フィルターノードの追加

1. **rbe**（report by exception）ノードをキャンバスにドラッグします。

2. ダブルクリックして設定を開きます。

   - **Mode**を**block unless value changes**に設定
   - **Property**を`msg.payload.temperature`に設定

3. **Done**をクリックします。

4. **json**ノードの出力を**rbe**ノードの入力に接続します。

このフィルターノードは、前回と同じ温度値の場合にメッセージをブロックし、繰り返しの同一データによる不要なトラフィックを削減します。

![フィルターノードの追加](./assets/add_filter_node.png)

### ステップ3: テンプレートノードの追加

1. **template**ノードをキャンバスにドラッグします。

2. ダブルクリックして設定を開き、Mustache構文を使って出力フォーマットを入力します。例：

   ```
   {"temperature": {{payload.temperature}}, "humidity": {{payload.humidity}}}
   ```

3. **Done**をクリックします。

4. **rbe**ノードの出力を**template**ノードの入力に接続します。

![テンプレートノードの追加](./assets/add_template_node.png)

### ステップ4: 出力ノードの接続とデプロイ

1. **template**ノードの出力を**mqtt-out**ノードの入力に接続します。

2. **Deploy**をクリックします。

3. すべてのノードが緑色の**connected**ステータスを表示していることを確認します。

> フィルタリングしたデータを再整形せずにそのままパブリッシュしたい場合は、**template**ノードを省略し、**rbe**ノードを直接**mqtt-out**に接続してください。

![高度なノードの接続](./assets/connect_advanced_nodes.png)

## フローのテスト

MQTTXや任意のMQTTクライアントを使ってパイプラインをテストします：

1. `test/node_red/out`をサブスクライブして処理結果を監視します。

2. `test/node_red/in`に以下のようなJSONペイロードをパブリッシュします：

   ```json
   {"temperature": 25, "humidity": 60}
   ```

3. 出力トピックにメッセージが現れることを確認します。

4. 同じメッセージを再度パブリッシュします。**rbe**フィルターにより重複が抑制され、出力は現れません。

5. 温度値を変更してパブリッシュします：

   ```json
   {"temperature": 26, "humidity": 60}
   ```

6. このメッセージはフィルターを通過し、出力トピックに表示されることを確認します。

![フローのテスト](./assets/test_the_flow.png)

## トラブルシューティング

### ノードが「disconnected」ステータスを表示する

**状況**

- デプロイ後、**mqtt-in**または**mqtt-out**ノードが赤い**disconnected**インジケーターを表示する。

**考えられる原因**

- ブローカーアドレスまたはポートの誤り
- ネットワークファイアウォールによるポート`1883`または`8883`のブロック
- ブローカーが起動していない

**対処法**

- ノードをダブルクリックし、**Server**横の鉛筆アイコンをクリックしてブローカーアドレスとポートを確認する。
- MQTTXなど別のMQTTクライアントでブローカーへの基本接続をテストする。
- TLSを使用している場合は、正しいポート（`8883`）とCA証明書が設定されているか確認する。

### 入力トピックでメッセージが受信されない

**状況**

- **mqtt-in**ノードは接続済みだがメッセージが届かない。

**考えられる原因**

- パブリッシャーとサブスクライバー間でトピック名が一致していない
- QoSレベルの不整合
- ブローカーのACLルールでサブスクライブがブロックされている

**対処法**

- パブリッシャーが**mqtt-in**ノードで設定したトピック（`test/node_red/in`）に正確に送信しているか確認する。
- Node-REDのデバッグノードを使い、フローの各段階でメッセージを検査する。
- ブローカーの認証およびACL設定を確認する。

### フィルターノードがすべてのメッセージをブロックする

**状況**

- 温度値が変わっても出力トピックにメッセージが現れない。

**考えられる原因**

- **rbe**ノードのプロパティパスが誤っている
- **json**ノードがフィルター前にペイロードをパースしていない

**対処法**

- **json**ノードが**rbe**ノードの前に配置され、**Always Convert to JavaScript Object**に設定されていることを確認する。
- **rbe**ノードのプロパティが`msg.payload.temperature`（`payload.temperature`ではない）に設定されているか確認する。
- **json**ノードの後に**debug**ノードを追加し、`msg.payload`の構造を検査する。

### 認証に失敗する

**状況**

- デプロイ直後にノードが**disconnected**となり、ブローカーのログに認証エラーが記録される。

**考えられる原因**

- ブローカー設定にユーザー名またはパスワードが未設定または誤っている
- トピックに対するACL制限

**対処法**

- ノードをダブルクリックし、ブローカー設定の**Security**タブで正しいユーザー名とパスワードを入力する。
- EMQXの認証設定を確認する。

## さらに詳しく

詳細な解説や追加の例については、ブログ記事[Using Node-RED to Process MQTT Data](https://www.emqx.com/en/blog/using-node-red-to-process-mqtt-data)をご覧ください。