Data Integrationを使った接続承認イベントトピックメッセージの取得

接続承認イベントトピック $events/client_connack は、サーバーがクライアントに CONNACK パケットを送信するたびに発生します。reason_code フィールドは接続の成功可否を示し、失敗した場合は具体的な原因を提供するため、クライアント接続問題の診断における重要な参照情報となります。通常のクライアントはこのトピックに直接サブスクライブできませんが、ルールエンジンを通じてこれらのイベントメッセージを取得し、データベースへの書き込みや他のトピックへの転送、リアルタイム監視・分析などの処理が可能です。

EMQXブローカーはMQTTブローカーとして機能し、ストレージサービスではないため、セキュリティおよびプライバシーの制約から、クライアントの過去の状態データやデバッグレベルのイベントログをデフォルトで保存しません。したがって、デバイスの接続状態をリアルタイムに把握し、トラブルシューティングや監査分析に不可欠な行動記録を保持するために、このイベントトピックを基にデータ統合ルールを設定することを推奨します。

本ページでは、接続承認イベントトピックのユースケースを紹介し、ルールエンジンを用いたイベントメッセージの取得・処理方法を解説するとともに、MQTTX Desktop を使ったこのトピックからのイベントメッセージ受信シミュレーション方法を示します。

ユースケース

接続承認イベントトピックは以下のようなシナリオで活用できます。

• 接続失敗の診断：reason_code や関連フィールドを取得・記録し、接続失敗の原因特定に役立てる。
• デバイスアクセス品質の監視：接続試行をリアルタイムで追跡し、失敗が多いデバイスや地域を特定、アクセス設定やネットワーク状況、認証方式の最適化を図る。
• セキュリティ監査およびアクセス追跡：接続イベントデータを外部プラットフォームに転送し、セキュリティ監査やコンプライアンス分析を支援する。

トリガー条件と主要フィールド

接続承認イベント ($events/client_connack)

• トリガー条件：サーバーがクライアントに CONNACK パケットを送信した際に発生。
• 主要フィールド：

フィールド名	説明
reason_code	接続成功または失敗原因を示す理由コード
clientid	接続してきたクライアントのID
username	認証時に使用されたユーザー名
peername	クライアントのIPアドレスとポート
keepalive	MQTTのキープアライブ間隔（クライアントとブローカー間のハートビート間隔）
clean_start	MQTTの clean_start フラグ（クライアントが前回のセッションを再利用するかどうか）
expiry_interval	MQTTセッションの有効期限（切断後にブローカーがセッションを保持する期間）
timestamp	イベント発生時のタイムスタンプ（ミリ秒単位）

TIP

接続承認イベントの全フィールド一覧は公式ドキュメントのデータソースとフィールドを参照してください。

接続失敗シナリオの解析

接続承認イベントの reason_code フィールドは接続エラーの詳細な理由を提供し、問題の診断やクライアントロジックの最適化に役立ちます。

MQTT v3.1.1で定義される主な理由コードは以下の通りです。

理由コード	説明
connection_accepted	クライアントが受理され、正常に接続された。
unacceptable_protocol_version	ブローカーがクライアントの要求するMQTTプロトコルバージョンをサポートしていない。
client_identifier_not_valid	クライアントIDはUTF-8準拠だが許可されていない（例：Clean Start=0で空のクライアントID、またはIDがブローカーの制限を超過）。
server_unavailable	ネットワークは確立されたが、ブローカーが一時的に利用不可（例：高負荷、認証バックエンド障害）。
malformed_username_or_password	ユーザー名またはパスワードが不正（無効な文字、エンコードエラー、欠落フィールドなど）。
unauthorized_client	無効な認証情報や認可されていない操作により、クライアントが接続を許可されていない。

MQTT 5.0では、より多くの理由コードが用意されており、より詳細な失敗診断が可能です。主な例は以下の通りです。

理由コード	説明
success	接続が成功した。
unspecified_error	他の理由コードに該当しない未特定のエラー。
malformed_packet	CONNECTパケットが不正（フィールド欠落、エンコード不正など）。
protocol_error	プロトコル違反や誤用（例：複数のCONNECTパケット送信）。
implementation_specific_error	パケットは有効だがブローカー実装により拒否された。
unacceptable_protocol_version	ブローカーがクライアントのプロトコルバージョンをサポートしていない。
client_identifier_not_valid	クライアントIDはUTF-8準拠だがブローカーにより拒否された。
bad_username_or_password	ユーザー名またはパスワードが無効。
not_authorized	クライアントが認可されていない（例：ユーザー名不一致）。
server_unavailable	ブローカーが一時的に利用不可。
server_busy	ブローカーが多忙で新規接続を受け入れられない。
banned	クライアントが禁止されている（例：異常行動、ブラックリスト）。
bad_authentication_method	認証方式がサポートされていない、または再認証時に変更された。
topic_name_invalid	トピック名は形式的に有効だが、ブローカーまたはクライアントにより拒否された。
packet_too_large	パケットサイズが制限を超過（例：Willメッセージが大きすぎる）。
quota_exceeded	クライアントがブローカーの接続クォータを超過した。
retain_not_supported	ブローカーがリテインメッセージをサポートしていないが、Willメッセージがリテイン指定されている。
qos_not_supported	クライアントがブローカー非対応のQoSレベルを要求した。
use_another_server	ブローカーがクライアントに一時的に別サーバーへの切り替えを指示。
server_moved	ブローカーがクライアントに恒久的な別サーバーへの切り替えを指示。
connection_rate_exceeded	接続頻度がブローカーの制限を超過（例：頻繁な再接続）。

データ統合でイベントトピックメッセージを取得する設定

実際には、イベントトピックのメッセージは主に以下の2つの方法で処理されます。

1. メッセージの再パブリッシュ：イベントメッセージを別のMQTTトピックに転送します。軽量かつリアルタイムで、MQTTエコシステムとシームレスに連携可能です。
2. 外部サービスへの転送：イベントメッセージをデータベースやメッセージキュー、HTTPサービスなどの外部システムに送信します。永続的な保存や下流システムとの統合が可能です。

本ページでは、メッセージの再パブリッシュとHTTPサービスへの転送を例示します。その他の統合については、EMQX Cloud データ統合を参照してください。

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

接続承認イベントメッセージを別トピックに再パブリッシュする方法を示します。

ルールとアクションの作成

1. データ統合ページのデータ転送でRepublishを選択します。既にコネクターがある場合はコネクター作成をクリックし、Republishを選びます。

2. SQLエディターでルールSQLを定義します。接続失敗診断用の例は以下の通りです。

   SELECT
       clientid,
       username,
       reason_code,
       timestamp
   FROM
       "$events/client_connack"

3. 次へをクリックしてアクション追加画面へ進みます。

4. 新規アクションで以下を設定します。

   • コネクター：デフォルトの Republish を使用。
   • トピック：転送先トピックを client_connack に設定。
   • ペイロード、QoS、Retain：デフォルトのまま。

5. 確認をクリックして完了します。

ルールとアクションのテスト

MQTTX を使ってクライアント接続をシミュレートすることを推奨しますが、任意のMQTTクライアントでも構いません。

1. MQTTXでClientID sub のクライアントを接続します。
2. 再パブリッシュ先トピック client_connack をサブスクライブします。
3. 別の接続 conn を作成すると、sub クライアントは以下のようなメッセージを受信します。

{
    "clientid": "conn",
    "username": "u_emqx",
    "reason_code": "success",
    "timestamp": 1645003578856
}

イベントトピックメッセージをHTTPサービスに転送

接続承認イベントメッセージをHTTPサービスに転送する方法を示します。開始前に、コネクターへ内部IPでアクセスするためのVPCピアリング接続を作成するか、パブリックIPでアクセスするためのNATゲートウェイを有効にしてください。

HTTPサーバーコネクターの作成

1. データ統合のWebサービスでHTTPサーバーを選択します。既にコネクターがある場合はコネクター作成をクリックし、HTTPサーバーを選びます。

2. URLを入力し、必要に応じて他の設定を調整します。

   URLは転送先のHTTPサービスを指し、コネクターはルールで定義されたペイロードをPOSTリクエストで送信します。

3. テストをクリックして、EMQXが指定URLに正常に到達できるか確認します。

4. 新規作成をクリックしてコネクター作成を完了します。

ルールとアクションの作成

1. ルール作成をクリックして開始します。

2. SQLエディターで以下のようなSQLを定義します。

   SELECT
       clientid,
       username,
       reason_code,
       timestamp
   FROM
       "$events/client_connack"

3. 次へをクリックしてアクション作成へ進みます。

4. 先ほど作成したHTTPサーバーコネクターを選択し、その他の設定はデフォルトのままにします。

5. 確認をクリックしてルール作成を完了します。

TIP

MQTTデータをHTTPサービスに転送する全体のワークフローは、MQTTデータをHTTPサーバーに取り込むを参照してください。

ルールとアクションのテスト

1. MQTTXでClientID conn で接続します。
2. HTTPサービスは以下のようなペイロードを含むPOSTリクエストを受信します。

{
    "clientid": "conn",
    "username": "u_emqx",
    "reason_code": "success",
    "timestamp": 1645003578856
}