EMQX Cloud

日本語

简体中文
English

日本語

简体中文
English

Appearance

お問い合わせ
お問い合わせ

メッセージの再パブリッシュ ​

Data Integrationを通じて、特定の条件を満たすメッセージをコードを書くことなく他のトピックに再パブリッシュすることが可能です。EMQXブローカーでは、ルールを作成し、ルールSQLでソースメッセージのデータをフィルタリングおよび処理し、「メッセージ再パブリッシュ」アクションをルールに追加して、処理結果をメッセージパブリッシュで転送できます。

本ページでは、任意のメッセージのmsgに文字列helloが含まれるたびに、そのメッセージをgreetトピックに再パブリッシュするData Integrationの作成方法をデモンストレーションします。主な手順は以下の通りです。

  1. フィルタリング条件を設定するルールの作成。
  2. ルールにメッセージ再パブリッシュのアクションを追加。
  3. Data Integrationの作成完了とテストの実施。

Data Integrationによるメッセージ再パブリッシュの設定は、コネクターの追加を必要としません。以下のセクションで具体的な設定手順を説明します。

TIP

ルールのマッチとメッセージ再パブリッシュアクションの実行はData IntegrationのTPSにカウントされます。アクションによって生成されたメッセージもデプロイメントのメッセージ受信TPSにカウントされます。再パブリッシュされたメッセージがサブスクライブしているクライアントに送信される場合、その配信はメッセージ送信TPSにカウントされます。

ルールの作成 ​

  1. Data Integration ページの Data Forward サービスカテゴリで Republish をクリックします。既に他のコネクターを作成している場合は、New Connector をクリックし、Data Forward サービスカテゴリの下にある Republish を選択します。

  2. SQL Editor にルールSQLを定義し、任意のメッセージのmsgに文字列helloが含まれる場合にエンジンをトリガーするようにします。

    • FROM句でメッセージデータのソースを指定します。このデモではすべてのトピック（#）のメッセージを対象とします。
    • WHERE句でメッセージペイロード内のmsgに対して正規表現マッチングを行い、helloを含む場合にData Integrationを実行します。

    上記の原則に基づくSQL定義の例は以下の通りです。

    sql
    SELECT
  payload.msg as msg
FROM
  "#"
WHERE  
  regex_match(msg, 'hello')

  3. SQL入力ボックス下の Try It Out をクリックし、以下のデータを入力してみてください。

    • Topic: t/a
    • Payload:
    json
    {
  "msg": "hello test"
}

    Test をクリックし、Output Result を確認します。正しく設定されていれば、フィールドと値に完全なJSONデータが表示されます。

    json
    {
  "msg": "hello test"
}

    テスト出力が期待通りであれば、次のステップに進めます。

    注意: テストに失敗した場合は、SQLの記述が正しいかご確認ください。

アクションの追加 ​

  1. New Rule ステップページで Next をクリックし、アクションを追加します。

  2. New Action ステップページで以下のパラメータを設定します。

    • Connector: デフォルトの Republish のままにします。
    • Topic: 送信先トピックを greet に設定します。
    • Payload: メッセージ内容のテンプレートとして ${msg} -- forward from EMQX Cloud を入力します。
    • QoSはデフォルト値のままにします。

  3. 必要に応じて、トグルスイッチをクリックして MQTT 5.0 Message Properties オプションを設定します。詳細はMQTT 5.0 Message Propertiesをご参照ください。

  4. Direct Dispatch を有効にする場合はトグルスイッチをクリックしてください。有効化するとメッセージは直接サブスクライバーに配信され、追加のルールトリガーや同一ルールの再帰的な発動を防止します。

  5. Confirm をクリックしてアクションとルールの作成を完了します。

  6. Successful new rule ポップアップで Back to Rules をクリックし、Test Message Republishの手順に従ってルールをテストします。あるいは、Test Rules をクリックしてページ上でシミュレーションデータを入力しルールをテストすることも可能です。詳細はTest Rulesをご参照ください。

メッセージ再パブリッシュのテスト ​

MQTTXを使ってメッセージ送信をシミュレートすることを推奨しますが、他の任意のクライアントでも構いません。

  1. MQTTXでデプロイメントに接続し、testトピックに以下のメッセージを送信します。

    json
    {
  "msg": "hello"
}

  2. ルール一覧からメッセージ再パブリッシュルールを探し、ルールIDをクリックしてルール統計ページに入ります。関連する統計指標が表示されます。リセットボタンをクリックしてメトリクスデータをリセットできます。

    TIP

    サーバレスデプロイメントではメトリクスのリセットはサポートされていません。

    rule_04

  3. クライアントで greet トピックをサブスクライブします。msgにhelloが含まれている場合はメッセージが転送され、含まれていない場合は転送されないことを確認できます。

    Republishing

アクションパラメータ ​

パラメータ説明
Topic再パブリッシュするメッセージの送信先トピックを設定します。${field}構文による動的トピック構築をサポートします。入力欄に$を入力すると利用可能な変数のドロップダウンリストが表示されます。
QoS再パブリッシュするメッセージのQoSレベルを設定します。
Retainこのメッセージをリテインメッセージとして転送するかどうかを設定します。
Payload再パブリッシュするメッセージの本文内容を設定します。ルールSQLの出力フィールドを参照するために${field}構文をサポートします。エディターに$を入力すると利用可能な変数のドロップダウンリストが表示されます。${payload}を入力すると元のメッセージと同じペイロードを変更せずに再パブリッシュします。
MQTT 5.0 Message Propertiesトグルスイッチをクリックして必要に応じてメッセージプロパティを設定します。再パブリッシュするメッセージにリッチなメッセージメタデータを付加できます。詳細は下記参照。
Direct Dispatch有効にするとメッセージは直接サブスクライバーに配信され、追加のルールトリガーや同一ルールの再帰的発動を防止します。

MQTT 5.0 メッセージプロパティ ​

パラメータ説明
Payload Format Indicatorペイロードのフォーマットを示します。falseに設定するとメッセージは未判定のバイト列として扱われます。trueに設定するとペイロードはUTF-8エンコードされた文字データとして扱われ、クライアントが内容を効率的に解析できます。
Message Expiry Intervalメッセージが配信先に届かない場合にいつ期限切れとするかを秒単位で指定します。
Content Type再パブリッシュするメッセージのペイロード内容の種類やフォーマット（MIMEタイプ）を指定します。例：text/plain、audio/aac、application/jsonなど。
Response Topic応答メッセージをパブリッシュする特定のMQTTトピックを入力します。例：response/my_device。
Correlation Data応答を元のリクエストに紐付けるための一意の識別子を入力します。トランザクションIDやリクエスト識別子などが該当します。