GCP Pub/Sub への MQTT データ取り込み

Google Cloud Pub/Sub は、非常に高い信頼性とスケーラビリティを実現するために設計された非同期メッセージングサービスです。EMQX は、MQTT データのリアルタイム抽出、処理、分析のために Google Cloud Pub/Sub とのシームレスな統合をサポートしています。Cloud Functions、App Engine、Cloud Run、Kubernetes Engine、Compute Engine などのさまざまな Google Cloud サービスへデータをプッシュできます。また、Google Cloud から MQTT へのデータ配信も可能で、ユーザーが GCP 上で迅速に IoT アプリケーションを構築できるよう支援します。

本ページでは、EMQX と GCP Pub/Sub 間のデータ統合について包括的に紹介し、データ統合の作成と検証に関する実践的な手順を提供します。

動作概要

GCP Pub/Sub データ統合は、EMQX の標準機能であり、MQTT データストリームを Google Cloud とシームレスに連携させ、豊富なサービスと機能を活用して IoT アプリケーション開発を支援します。

GCP_bridge_architect

EMQX はルールエンジンと Sink を介して MQTT データを GCP Pub/Sub に転送します。GCP Pub/Sub プロデューサーの例で、全体の流れは以下の通りです。

IoT デバイスがメッセージをパブリッシュ：デバイスは特定のトピックを通じてテレメトリやステータスデータをパブリッシュし、ルールエンジンをトリガーします。
ルールエンジンがメッセージを処理：組み込みのルールエンジンを用いて、特定のソースからの MQTT メッセージをトピックマッチングに基づき処理します。ルールエンジンは対応するルールをマッチさせ、データフォーマットの変換、特定情報のフィルタリング、コンテキスト情報の付加などを行います。
GCP Pub/Sub へのブリッジング：ルールはメッセージを GCP Pub/Sub に転送するアクションをトリガーし、データプロパティ、オーダーキー、MQTT トピックと GCP Pub/Sub トピックのマッピングを簡単に設定できます。これにより、より豊かなコンテキスト情報と順序保証が提供され、柔軟な IoT データ処理が可能となります。

MQTT メッセージデータが GCP Pub/Sub に書き込まれた後は、以下のような柔軟なアプリケーション開発が可能です。

リアルタイムデータ処理と分析：Dataflow、BigQuery、Pub/Sub のストリーミング機能など、強力な Google Cloud のデータ処理・分析ツールを活用し、メッセージデータのリアルタイム処理と分析を行い、貴重なインサイトや意思決定支援を得られます。
イベント駆動型機能：Cloud Functions や Cloud Run などの Google Cloud イベント処理をトリガーし、動的かつ柔軟な関数の起動と処理を実現します。
データ保存と共有：Cloud Storage や Firestore などの Google Cloud ストレージサービスにメッセージデータを送信し、大量データの安全な保存と管理を行います。これにより、他の Google Cloud サービスと連携してデータの共有や分析が可能となり、多様なビジネスニーズに対応します。

特長とメリット

GCP Pub/Sub とのデータ統合は以下の特長とメリットを提供します。

堅牢なメッセージングサービス：EMQX と GCP Pub/Sub は共に高可用性とスケーラビリティを備え、大規模なメッセージストリームの確実な受信、配信、処理を保証します。IoT データの順序維持、メッセージ品質保証、パーシステンス（永続化）をサポートし、信頼性の高いメッセージ伝送と処理を実現します。
柔軟なルールエンジン：組み込みのルールエンジンにより、特定のソースメッセージやイベントをトピックマッチングで処理可能です。データフォーマット変換、特定情報のフィルタリング、コンテキスト情報の付加などが行え、GCP Pub/Sub と組み合わせてさらなる処理や分析が可能です。
豊富なコンテキスト情報：GCP Pub/Sub データ統合を通じて、メッセージにより豊かなコンテキスト情報を付加できます。クライアント属性を Pub/Sub 属性やソートキーにマッピングすることで、後続のアプリケーション開発やデータ処理においてより正確な分析と処理を支援します。

まとめると、EMQX と GCP Pub/Sub の統合により、高信頼かつスケーラブルなメッセージ配信と、データ分析・統合のための豊富なツールとサービスが利用可能となります。これにより、堅牢な IoT アプリケーションの構築と、イベント駆動型の柔軟なビジネスロジックの実装が可能となります。

はじめる前に

このセクションでは、GCP Pub/Sub データ統合の作成を開始する前に必要な準備について説明します。

前提条件

EMQX データ統合のルールに関する知識
データ統合に関する知識

GCP でサービスアカウントキーを作成する

Service Account JSON 認証を使用する場合は、GCP でサービスアカウントを作成し、JSON 形式のキーを生成してください。

GCP アカウントでサービスアカウントを作成します。サービスアカウントには、対象トピック（例：Pub/Sub Editor ロール）へのメッセージの検査/読み取りおよびパブリッシュ権限が必要です。
作成したサービスアカウントのメールアドレスをクリックし、Key タブを選択します。Add key のドロップダウンリストから Create new key を選択し、サービスアカウントキーを作成して JSON 形式でダウンロードします。
TIP
サービスアカウントキーは安全に保管し、後で使用します。

GCP で Workload Identity Federation を設定する

Workload Identity Federation（WIF）を利用すると、EMQX は長期間有効なサービスアカウントキーを使用せずに GCP リソースにアクセスできます。EMQX は外部 ID プロバイダー（例：Microsoft Azure）からトークンを取得し、GCP の Security Token Service を通じて一時的な GCP トークンと交換し、そのトークンでサービスアカウントを代行します。トークンの更新は自動で行われます。

WIF を使用するには、コネクター作成前に GCP プロジェクトで以下を完了してください。

Google Cloud コンソールで IAM & Admin -> Workload Identity Federation に移動し、ワークロードアイデンティティプールを作成し、Pool ID と Project Number を控えます。
プールにプロバイダーを追加し、Provider ID を控えます。OIDC ベースの認証の場合、外部 ID プロバイダーから OAuth 2.0 クライアント資格情報（クライアント ID、クライアントシークレット、トークンエンドポイント URI）を取得します。
ワークロードアイデンティティプールに、Pub/Sub トピックにアクセス可能な GCP サービスアカウントの代行権限を付与します。コネクター設定時にサービスアカウントのメールアドレスが必要です。
TIP
詳細はWorkload Identity Federation の設定を参照してください。

例：Microsoft Azure (Entra ID)

Microsoft Entra IDで API を公開するアプリケーションを登録し、クライアントシークレットを作成します。コネクター設定時に以下の値を使用します。

コネクター項目	値
Endpoint URI	`https://login.microsoftonline.com/<tenant-id>/oauth2/v2.0/token`
OAuth Client ID	アプリケーション（クライアント）ID、形式は `api://<application-id>`
OAuth Client Secret	アプリケーション用に生成したクライアントシークレット
OAuth Request Scope	`api://<application-id>/.default`

注意

scope はアプリケーションのオーディエンス（aud）と完全に一致させる必要があります。そうでないと GCP STS とのトークン交換に失敗します。詳細は Microsoft ドキュメントのOAuth 2.0 クライアント資格情報フローを参照してください。

サービスアカウントに WIF プールへのアクセス権を付与する際は、Application ID ではなく Object ID を Subject 値として使用してください。Object ID は Azure ポータルのアプリケーション概要ページの Enterprise applications で確認できます。

Attached Service Account の前提条件

Attached Service Account 認証を使用する場合、EMQX はサービスアカウントがアタッチされた GCP Compute Engine インスタンス上で動作している必要があります。インスタンスの OAuth アクセススコープが Pub/Sub へのアクセスを許可していることを確認してください。Google は cloud-platform スコープ（https://www.googleapis.com/auth/cloud-platform）の使用を推奨し、サービスアカウントの権限は IAM ロールで制限することを推奨しています。サービスアカウントは対象の Pub/Sub トピックおよびサブスクリプションへのアクセス権限を持つ必要があります。詳細は Google Cloud ドキュメントのサービスアカウントを参照してください。

対象の Pub/Sub トピックとサブスクリプションは、Compute Engine インスタンスに関連付けられた GCP プロジェクト内に存在する必要があります。EMQX クラスターの場合、すべてのノードがこれらの要件を満たし、該当プロジェクトの Compute Engine インスタンス上で動作している必要があります。

コネクター起動時に、EMQX はインスタンスメタデータエンドポイントから GCP プロジェクト ID とアクセストークンを自動取得します。サービスアカウントキーのアップロードは不要です。

GCP でトピックの作成と管理

EMQX で GCP Pub/Sub データ統合を設定する前に、トピックを作成し、GCP での基本的な管理操作に慣れておく必要があります。

Google Cloud コンソールで Pub/Sub -> Topics ページに移動します。詳細はトピックの作成と管理を参照してください。
TIP
サービスアカウントにそのトピックへのパブリッシュ権限が必要です。
Topic ID フィールドにトピックの ID を入力し、Create topic をクリックします。
Subscriptions ページに移動し、一覧から該当の Topic ID をクリックします。トピックに対してサブスクリプションを作成します。
- Delivery type で Pull を選択します。
- Message retention duration は 7 日を選択します。
詳細はGCP Pub/Sub サブスクリプションを参照してください。
Subscription ID -> Messages -> Pull をクリックすると、トピックに送信されたメッセージを確認できます。

GCP Pub/Sub プロデューサーコネクターの作成

GCP Pub/Sub プロデューサー Sink アクションを追加する前に、EMQX と GCP Pub/Sub 間の接続を確立するための GCP Pub/Sub プロデューサーコネクターを作成します。

EMQX ダッシュボードで Integration -> Connector をクリックします。
ページ右上の Create をクリックし、コネクター選択画面で Google PubSub Producer を選択して Next をクリックします。
名前と説明を入力します。例：my-pubsubproducer。名前は GCP Pub/Sub プロデューサー Sink とコネクターを関連付けるために使用され、クラスター内で一意である必要があります。
Authentication リストから以下のいずれかの認証方法を選択し、対応するフィールドを設定します。
- Service Account JSON：GCP でサービスアカウントキーを作成するでエクスポートした JSON 形式のサービスアカウント認証情報をアップロードします。
- Workload Identity Federation (WIF)：以下のフィールドを入力します。前提条件はSet Up Workload Identity Federation in GCPを参照してください。
  - GCP Project ID：コネクターがアクセスするリソースのプロジェクト ID。
  - GCP Project Number：コネクターがアクセスするリソースのプロジェクト番号。
  - Service Account Email：代行するサービスアカウントのメールアドレス。
  - Workload Identity Pool ID：WIF トークン交換に使用するワークロードアイデンティティプールの ID。
  - Workload Identity Provider ID：WIF トークン交換に使用するワークロードアイデンティティプロバイダーの ID。
  - Initial Token Configuration で認証タイプを選択し、対応するフィールドを入力します。現在サポートされているのは OIDC with Client Credentials Grant Type のみです。
    - Endpoint URI：OIDC プロバイダーの OAuth トークンエンドポイント URI。
    - OAuth Client ID：OAuth サーバーからトークンを要求するためのクライアント ID。
    - OAuth Client Secret：OAuth サーバーからトークンを要求するためのクライアントシークレット。
    - OAuth Request Scope：OAuth アクセストークンを要求する際に指定する scope（プロバイダーによって必要な場合）。
- Attached Service Account：追加のフィールドは不要です。EMQX はインスタンスメタデータエンドポイントから GCP プロジェクト ID とアクセストークンを自動取得します。前提条件はAttached Service Account の前提条件を参照してください。
Create をクリックする前に、Test Connectivity をクリックしてコネクターが GCP Pub/Sub サーバーに接続できるかテストできます。
ページ下部の Create ボタンをクリックしてコネクター作成を完了します。ポップアップダイアログで Back to Connector List をクリックするか、Create Rule をクリックして GCP Pub/Sub プロデューサー Sink を使ったルール作成に進めます。詳細はCreate a Rule with GCP Pub/Sub Producer Sinkを参照してください。

GCP Pub/Sub プロデューサー Sink を使ったルールの作成

このセクションでは、GCP Pub/Sub に保存するデータを指定するルールの作成方法を説明します。

EMQX ダッシュボードで Integration -> Rules をクリックします。
ページ右上の Create をクリックします。
ルール ID に my_rule と入力します。
SQL Editor にルールを設定します。例えば、トピック /devices/+/events の MQTT メッセージを GCP Pub/Sub に保存したい場合、以下の SQL を使用します。
注意：独自の SQL を指定する場合、Sink のペイロードテンプレートで必要なすべてのフィールドが SELECT 部分に含まれていることを確認してください。
sql
```
SELECT
  *
FROM
  "/devices/+/events"
```
注意：初心者の方は SQL Examples と Enable Test をクリックして SQL ルールの学習とテストが可能です。
+ Add Action ボタンをクリックして、ルールでトリガーされるアクションを定義します。Type of Action ドロップダウンリストから Google PubSub Producer を選択し、EMQX がルールで処理したデータを GCP Pub/Sub に送信するようにします。
Action ドロップダウンは Create Action のままにするか、既存の GCP Pub/Sub プロデューサー Sink を選択できます。この例では新規 Sink を作成してルールに追加します。
Name フィールドに Sink の名前を入力します。名前は英数字の組み合わせとしてください。
Connector ドロップダウンから先ほど作成した my_pubsubprodcer を選択します。隣のボタンから新規コネクター作成も可能です。設定パラメータはCreate a Connectorを参照してください。
GCP PubSub Topic に以下のいずれかを入力します。
- 例：my-iot-core のように、GCP でトピックの作成と管理で作成したトピック名。EMQX は設定したサービスアカウントのプロジェクト内でトピックを解決します。
- projects/<project-id>/topics/<topic-name> の形式の完全修飾トピックパス。異なる GCP プロジェクトのトピックにパブリッシュする場合に使用します。該当プロジェクトのトピックに対してサービスアカウントに必要な Pub/Sub 権限を付与してください。
Payload Template にテンプレートを定義するか空欄のままにします。
- 空欄の場合、MQTT メッセージの clientid、topic、payload などの可視入力を JSON 形式でエンコードします。
- 定義したテンプレートを使う場合、${variable_name} 形式のプレースホルダーが MQTT コンテキストの対応値で置換されます。例：${topic} は MQTT メッセージのトピック my/topic に置換されます。
Attributes Template と Ordering Key Template に、送信メッセージの属性やオーダーキーのフォーマットテンプレートを定義します（任意）。
- Attributes はキーと値の両方に ${variable_name} 形式のプレースホルダーを使え、MQTT コンテキストから値を抽出します。キーのテンプレートが空文字になる場合、そのキーは GCP Pub/Sub メッセージに含まれません。
- Ordering Key は ${variable_name} 形式のプレースホルダーを使えます。解決結果が空文字の場合、GCP Pub/Sub メッセージの orderingKey フィールドは設定されません。
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義できます。詳細はフォールバックアクションを参照してください。
Advanced Settings を展開して必要に応じてオプション設定を行います。詳細はAdvanced Settingsを参照してください。
Create をクリックする前に、Test Connectivity をクリックしてコネクターが GCP Pub/Sub サーバーに接続できるかテストできます。
Create ボタンをクリックして Sink 設定を完了すると、新しい Sink が Action Outputs タブに表示されます。
Create Rule ページに戻り、Create をクリックしてルールを作成します。

これでルールが正常に作成されました。Integration -> Rules ページで新規ルールを確認できます。Actions(Sink) タブをクリックすると、新しい Google PubSub Producer Sink が表示されます。

また、Integration -> Flow Designer をクリックするとトポロジーが表示され、トピック /devices/+/events のメッセージがルール my_rule によって解析され、GCP Pub/Sub に送信・保存されていることが視覚的に確認できます。

プロデューサールールのテスト

MQTTX を使ってトピック /devices/+/events にメッセージを送信します。
bash
```
mqttx pub -i emqx_c -t /devices/+/events -m '{ "msg": "hello GCP PubSub" }'
```
Sink の稼働状況を確認し、新規の受信メッセージと送信メッセージがそれぞれ 1 件ずつあることを確認します。
GCP コンソールの Pub/Sub -> Subscriptions に移動し、MESSAGES タブをクリックするとメッセージが確認できます。

GCP Pub/Sub コンシューマーコネクターの作成

GCP Pub/Sub コンシューマー Source を追加する前に、EMQX と GCP Pub/Sub 間の接続を確立するための GCP Pub/Sub コンシューマーコネクターを作成します。

EMQX ダッシュボードで Integration -> Connector をクリックします。
ページ右上の Create をクリックし、コネクター選択画面で Google PubSub Consumer を選択して Next をクリックします。
名前と説明を入力します。例：my-pubsubconsumer。名前は GCP Pub/Sub コンシューマー Sink とコネクターを関連付けるために使用され、クラスター内で一意である必要があります。
Authentication リストから以下のいずれかの認証方法を選択し、対応するフィールドを設定します。
- Service Account JSON：GCP でサービスアカウントキーを作成するでエクスポートした JSON 形式のサービスアカウント認証情報をアップロードします。
- Workload Identity Federation (WIF)：以下のフィールドを入力します。前提条件はSet Up Workload Identity Federation in GCPを参照してください。
  - GCP Project ID：コネクターがアクセスするリソースのプロジェクト ID。
  - GCP Project Number：コネクターがアクセスするリソースのプロジェクト番号。
  - Service Account Email：代行するサービスアカウントのメールアドレス。
  - Workload Identity Pool ID：WIF トークン交換に使用するワークロードアイデンティティプールの ID。
  - Workload Identity Provider ID：WIF トークン交換に使用するワークロードアイデンティティプロバイダーの ID。
  - Initial Token Configuration で認証タイプを選択し、対応するフィールドを入力します。現在サポートされているのは OIDC with Client Credentials Grant Type のみです。
    - Endpoint URI：OIDC プロバイダーの OAuth トークンエンドポイント URI。
    - OAuth Client ID：OAuth サーバーからトークンを要求するためのクライアント ID。
    - OAuth Client Secret：OAuth サーバーからトークンを要求するためのクライアントシークレット。
    - OAuth Request Scope：OAuth アクセストークンを要求する際に指定する scope（プロバイダーによって必要な場合）。
- Attached Service Account：追加のフィールドは不要です。EMQX はインスタンスメタデータエンドポイントから GCP プロジェクト ID とアクセストークンを自動取得します。前提条件はAttached Service Account の前提条件を参照してください。
Create をクリックする前に、Test Connectivity をクリックしてコネクターが GCP Pub/Sub サーバーに接続できるかテストできます。
ページ下部の Create ボタンをクリックしてコネクター作成を完了します。ポップアップダイアログで Back to Connector List をクリックするか、Create Rule をクリックして GCP Pub/Sub コンシューマー Source を使ったルール作成に進めます。詳細はCreate a Rule with GCP Pub/Sub Consumer Sourceを参照してください。

GCP Pub/Sub コンシューマー Source を使ったルールの作成

このセクションでは、GCP Pub/Sub からメッセージを消費し、EMQX に転送するルールの作成方法を説明します。Google PubSub Consumer Source を作成・設定し、ルールのデータ入力として追加します。また、Republish アクションを追加して GCP Pub/Sub から EMQX へのメッセージ転送を行います。

EMQX ダッシュボードで Integration -> Rules をクリックします。
ページ右上の Create をクリックします。
ルール ID に my_rule_source と入力します。
右側の Data Inputs タブでデフォルトの Input Messages を削除し、Add Input をクリックします。
Input Type ドロップダウンから Google PubSub Consumer を選択します。
Source ドロップダウンはデフォルトの Create Source のままにします。この例では新規 Source を作成してルールに追加します。
Source の Name と任意で Description を入力します。名前は英数字の組み合わせとし、例：my-gcppubsub-source。
Connector ドロップダウンから先ほど作成した my_pubsubconsumer を選択します。隣のボタンから新規コネクター作成も可能です。設定パラメータはCreate a Connectorを参照してください。
GCP Pub/Sub から EMQX へメッセージを消費するために、以下の情報を設定します。
- GCP PubSub Topic：トピック名（例：my-iot-core）または完全修飾トピックパス（projects/<project-id>/topics/<topic-name>）を入力します。トピック名は設定したサービスアカウントのプロジェクト内で解決されます。異なる GCP プロジェクトのトピックから消費する場合は完全修飾パスを入力し、サービスアカウントに該当トピックの Pub/Sub 権限を付与してください。コンシューマーサブスクリプションはサービスアカウントのプロジェクトに作成され、トピック参照のみ別プロジェクトを指します。
- Maximum Messages to Pull：1 回のプルリクエストで GCP Pub/Sub から取得する最大メッセージ数を指定します。実際の取得数は指定値未満の場合があります。
Advanced Settings を展開して必要に応じてオプション設定を行います。詳細はAdvanced Settingsを参照してください。
Create をクリックする前に、Test Connectivity をクリックして GCP Pub/Sub サーバーへの接続が成功するかテストできます。

Create をクリックして Source 作成を完了します。Source はルールの Data Inputs タブに追加され、SQL Editor のルールは以下のようになります。

sql

SELECT
  *
FROM
  "$bridges/gcppubsub:my-gcppubsub-source"

注意：初心者の方は SQL Examples と Enable Test をクリックして SQL ルールの学習とテストが可能です。

my-gcppubsub-source からの GCP Pub/Sub メッセージは、以下の GCP Pub/Sub から MQTT トピックへのマッピングテーブルに示すフィールドにアクセスできます。ルール SQL を調整してデータ処理が可能です。この例ではデフォルトの SQL を使用します。

フィールド名	説明
`attributes`	（任意）文字列のキー・バリューのペアを含むオブジェクト（存在する場合）
`message_id`	GCP Pub/Sub がこのメッセージに割り当てたメッセージ ID
`ordering_key`	（任意）メッセージの順序付けキー（存在する場合）
`publishing_time`	GCP Pub/Sub によって定義されたメッセージのタイムスタンプ
`topic`	発信元の GCP Pub/Sub トピック
`value`	（任意）メッセージのペイロード（存在する場合）

注意：各 GCP Pub/Sub から MQTT トピックへのマッピングは、ユニークな GCP Pub/Sub トピック名を含む必要があります。つまり、同じ GCP Pub/Sub トピックが複数のマッピングに存在してはなりません。

これで GCP Pub/Sub コンシューマー Source の作成は完了しましたが、メッセージはまだ EMQX に直接パブリッシュされません。次に、Add Republish Action to the Rule の手順に従い、Republish アクションを作成してルールに追加してください。

ルールに Republish アクションを追加する

このセクションでは、GCP Pub/Sub コンシューマー Source から消費したメッセージを転送し、EMQX トピック t/1 にパブリッシュするための Republish アクションの追加方法を説明します。

ページ右側の Action Output タブを選択し、Add Action ボタンをクリックします。Type of Action ドロップダウンから Republish アクションを選択します。
メッセージ再パブリッシュの設定を入力します。
- Topic：MQTT にパブリッシュするトピック。ここでは t/1 を入力します。
- QoS：0、1、2、または ${qos} を選択、もしくは他のフィールドから QoS を設定するためのプレースホルダーを入力します。${qos} を選択すると元のメッセージの QoS に従います。
- Retain：true または false を選択します。メッセージをリテインメッセージとしてパブリッシュするかどうかを決定します。プレースホルダーも使用可能です。この例では false を選択します。
- Payload：転送するメッセージペイロードのテンプレートを設定します。空欄の場合はルール出力結果をそのまま転送します。${.value} と入力すると GCP Pub/Sub メッセージのペイロードのみを転送します。
  MQTT ペイロードテンプレートのデフォルトは ${.} で、利用可能なすべてのデータを JSON オブジェクトとして含みます。例えば、すべてのオプションフィールドを含む GCP Pub/Sub メッセージの場合、以下のような JSON になります。
  json
```
{
  "attributes": {"attribute_key": "attribute_value"},
  "message_id": "1679665968238",
  "ordering_key": "my-ordering-key",
  "topic": "my-pubsub-topic",
  "publishing_time": "2023-08-18T14:15:18.470Z",
  "value": "my payload"
}
```
  GCP Pub/Sub メッセージのサブフィールドはドット表記でアクセス可能です。例：${.value} はメッセージのペイロード値に解決され、${.attributes.h1} は属性キー h1 の値に解決されます。存在しない値は空文字に置換されます。
- MQTT 5.0 メッセージプロパティ：デフォルトで無効です。詳細設定はRepublish アクションの追加を参照してください。
Create をクリックしてアクション作成を完了します。作成成功後、ルール作成ページに戻り、Republish アクションが Action Outputs タブに追加されます。
ルール作成ページで Create ボタンをクリックしてルール全体の作成を完了します。

これでルールが正常に作成されました。Rules ページで新規ルールを確認できます。Sources タブには新しく作成した GCP Pub/Sub コンシューマー Source が表示されます。

また、Integration -> Flow Designer をクリックしてトポロジーを表示できます。トポロジーを通じて、GCP Pub/Sub コンシューマー Source からのメッセージが Republish を経てトピック t/1 にパブリッシュされる様子を直感的に確認できます。

GCP Pub/Sub コンシューマールールのテスト

GCP Pub/Sub コンシューマー Source が GCP Pub/Sub からメッセージを消費し、EMQX の MQTT トピック t/1 に再パブリッシュすることを検証する手順は以下の通りです。

MQTTX CLI を使って EMQX の MQTT トピック t/1 をサブスクライブします。
bash
```
mqttx sub -t t/1 -v
```
Google Cloud コンソールで Pub/Sub -> Topics に移動し、my-iot-core トピックをクリックして以下のメッセージをパブリッシュします。
json
```
{"msg":"hello GCP PubSub"}
```
MQTTX がトピック t/1 で以下のメッセージを受信することを確認します。
text
```
topic: t/1
payload: {"msg":"hello GCP PubSub"}
```

詳細設定

このセクションでは、GCP Pub/Sub コネクター、プロデューサー Sink、コンシューマー Source の詳細設定について説明します。

コネクターの詳細設定

GCP Pub/Sub プロデューサーおよびコンシューマーコネクターは同じ詳細設定を使用します。

項目名	説明	デフォルト値
HTTP Pipelining	各レスポンスを待たずに送信できる最大 HTTP リクエスト数。`1` に設定すると次のリクエスト送信前にレスポンスを待ちます。	`100`
Connection Pool Size	コネクションプールに保持する接続数。	`8`
Connect Timeout	HTTP 接続確立の最大待機時間。	`15` 秒
Max Inactive	HTTP クライアントが再接続を試みるまでの最大非アクティブ時間。EMQX 6.0.4 以降、レスポンス待ちで 60 秒間リクエスト送信がない場合に再接続します。60 秒以上の値を設定すると優先されます。レスポンスが 60 秒以上かかる場合は、最長応答時間以上に設定してください。	`10` 秒
Max Retries	リクエスト送信エラー発生時の最大リトライ回数。	`2`
Start Timeout	コネクター作成後、正常化までの最大待機時間。	`5` 秒
Health Check Interval	コネクターのヘルスチェック間隔。	`15` 秒
Health Check Timeout	ヘルスチェック結果の最大待機時間。タイムアウトすると切断とみなされます。	`60` 秒

プロデューサー Sink とコンシューマー Source 共通の詳細設定

プロデューサー Sink とコンシューマー Source は以下の詳細設定を共有します。Health Check Interval のデフォルト値は異なります。

項目名	説明	プロデューサー Sink デフォルト	コンシューマー Source デフォルト
Request TTL	リクエストがバッファに入ってからレスポンスまたはアックを受け取るまでの最大時間。この間にレスポンスやアックがない場合、リクエストは期限切れとなります。	`45` 秒	`45` 秒
Health Check Interval	Sink または Source のヘルスチェック間隔。	`15` 秒	`30` 秒
Health Check Interval Jitter	複数のアクションやソースが同時にヘルスチェックを開始しないように、ヘルスチェック間隔に付加される一様ランダム遅延。	`0` ミリ秒	`0` ミリ秒
Health Check Timeout	ヘルスチェック結果の最大待機時間。タイムアウトすると切断とみなされます。	`60` 秒	`60` 秒

プロデューサー Sink 固有の詳細設定

GCP Pub/Sub プロデューサー Sink は以下の追加詳細設定を提供します。

項目名	説明	デフォルト値
Buffer Pool Size	GCP Pub/Sub 送信前にデータを格納・処理するバッファワーカーの数。	`16`
Dispatch Strategy	明示的なピックキーなしリクエストをバッファワーカーに割り当てる戦略。`Per Client ID` は同一クライアントのリクエストを同一ワーカーに割り当て、`Random` はワーカー間で分散。	`Per Client ID`
Max Buffer Queue Size	各バッファワーカーが保持可能な最大データ量。	`256` MB
Batch Size	1 バッチあたりの最大リクエスト数。`1` に設定するとバッチ処理を無効化。	`1`
Query Mode	リクエストを同期または非同期で送信するか制御。`Async` モードでは GCP Pub/Sub の応答を待たずにメッセージ処理を継続。	`Async`
Inflight Window	Query Mode が `Async` の場合、レスポンス待ちの最大リクエスト数。厳密な順序処理が必要な場合は `1` に設定。	`100`

コンシューマー Source 固有の詳細設定

GCP Pub/Sub コンシューマー Source は以下の追加詳細設定を提供します。

項目名	説明	デフォルト値
Ack Deadline	Source が配信済みメッセージをアックするまでの GCP Pub/Sub の待機時間の目安。期限切れ後はメッセージが再配信される可能性があります。サポート範囲は `10` ～ `600` 秒。	`60` 秒

JT/T 808 ゲートウェイ

GB/T 32960 ゲートウェイ

Volcengine RTC

シナリオ

GPT-Realtime

GCP Pub/Sub への MQTT データ取り込み ​

動作概要 ​

特長とメリット ​

はじめる前に ​

前提条件 ​

GCP でサービスアカウントキーを作成する ​

GCP で Workload Identity Federation を設定する ​

Attached Service Account の前提条件 ​

GCP でトピックの作成と管理 ​

GCP Pub/Sub プロデューサーコネクターの作成 ​

GCP Pub/Sub プロデューサー Sink を使ったルールの作成 ​

プロデューサールールのテスト ​

GCP Pub/Sub コンシューマーコネクターの作成 ​

GCP Pub/Sub コンシューマー Source を使ったルールの作成 ​

ルールに Republish アクションを追加する ​

GCP Pub/Sub コンシューマールールのテスト ​

詳細設定 ​

コネクターの詳細設定 ​

プロデューサー Sink とコンシューマー Source 共通の詳細設定 ​

プロデューサー Sink 固有の詳細設定 ​

コンシューマー Source 固有の詳細設定 ​