データソースとフィールド

EMQXのルールは、MQTTメッセージ、MQTTイベント、データブリッジなど、さまざまなデータソースからのデータを処理できます。

ルールエンジン構文のセクションで説明したように、FROM句でデータソースを指定し、対応するフィールドをSELECT句やwhere句で参照できます。本セクションでは、MQTTメッセージ、MQTTイベント、およびデータブリッジのフィールドについて紹介します。

MQTTメッセージ

EMQXのルールを使ってメッセージのパブリッシュを処理する場合、FROM句でメッセージのトピックを指定します。

例えば、以下のステートメントでは、トピックパターンt/#にパブリッシュされたメッセージから、payload.msg（AS句でmsgにリネーム）、clientid、username、payload、topic、qosのフィールドを選択しています。

例：

SELECT
  payload.msg as msg,
  clientid,
  username,
  payload,
  topic,
  qos
FROM
  "t/#"

出力例：

{
  "username": "u_emqx",
  "topic": "t/a",
  "qos": 1,
  "payload": "{\"msg\":\"hello\"}",
  "msg": "hello",
  "clientid": "c_emqx"
}

受信したMQTTメッセージから選択可能なフィールドは以下の通りです：

フィールド	説明
id	MQTTメッセージID
clientid	パブリッシャーのクライアントID
username	パブリッシャーのユーザー名
payload	MQTTペイロード
peerhost	クライアントのIPアドレス
topic	MQTTトピック
qos	QoSレベル
flags	フラグ
headers	メッセージ処理に関連する内部データ
pub_props	PUBLISHプロパティ（MQTT 5.0クライアントのみ）
timestamp	タイムスタンプ（単位：ミリ秒）
publish_received_at	PUBLISHメッセージがEMQXに到達した時刻（単位：ミリ秒）
node	イベントが発生したノード
client_attrs	クライアント属性

MQTTイベント

EMQXのルールを使って、クライアントのオンライン・オフラインやサブスクリプションなどのイベント通知を取得するためにイベントトピックからデータを抽出できます。イベントトピックは"$events/"で始まり、例えば"$events/client_connected"などがあり、ルールのFROM句で指定可能です。

TIP

デフォルトでは、クライアントはMQTTイベントメッセージを直接サブスクライブできません。本セクションではルールを使ってこれらのメッセージをサブスクライブする方法を説明します。MQTTイベントメッセージのデータはシステムトピックをサブスクライブすることでも取得可能です。

サポートされているイベントトピック一覧は以下の通りです。

イベントトピック一覧

イベントトピック名	説明
$events/message_delivered	メッセージ配信完了
$events/message_acked	メッセージ受領確認
$events/message_dropped	ルーティング時メッセージ破棄
$events/delivery_dropped	配信時メッセージ破棄
$events/client_connected	接続完了
$events/client_disconnected	切断
$events/client_connack	接続応答
$events/client_check_authz_complete	認可チェック完了
$events/client_check_authn_complete	認証チェック完了
$events/session_subscribed	サブスクライブ完了
$events/session_unsubscribed	サブスクライブ解除
$events/sys/alarm_activated	システムアラーム発動
$events/sys/alarm_deactivated	システムアラーム解除

メッセージ配信完了イベント ("$events/message_delivered")

このイベントトピックは、メッセージがクライアントに配信された際にルールをトリガーするために使用できます。

例えば、"$events/message_delivered"イベントトピックから、パブリッシャーのIDとユーザー名、メッセージトピック、メッセージのQoS、イベントが発生したEMQXノード、イベント発生時刻のフィールドを抽出するには、以下のステートメントを使用します。

例：

SELECT
  from_clientid,
  from_username,
  topic,
  qos,
  node,
  timestamp
FROM
  "$events/message_delivered"

出力例：

{
  "topic": "t/a",
  "timestamp": 1645002753259,
  "qos": 1,
  "node": "emqx@127.0.0.1",
  "from_username": "u_emqx_1",
  "from_clientid": "c_emqx_1"
}

各フィールドの詳細は以下の通りです。

コード	説明
id	MQTTメッセージID
from_clientid	パブリッシャーのクライアントID
from_username	パブリッシャーのユーザー名
clientid	サブスクライバーのクライアントID
username	サブスクライバーのユーザー名
payload	MQTTペイロード
peerhost	クライアントのIPアドレス
topic	MQTTトピック
qos	QoSレベル
flags	フラグ
pub_props	PUBLISHプロパティ（MQTT 5.0クライアントのみ）
timestamp	イベント発生時刻（単位：ミリ秒）
publish_received_at	PUBLISHメッセージがEMQXに到達した時刻（単位：ミリ秒）
node	イベントが発生したEMQXノード

メッセージ受領確認イベント ("$events/message_acked")

このイベントトピックは、メッセージ配信の受領確認があった際にルールをトリガーするために使用できます。

TIP

QoS 1およびQoS 2メッセージのみ対応しています。

例えば、"$events/message_acked"イベントトピックから、パブリッシャーのIDとユーザー名、メッセージトピック、メッセージQoS、イベントが発生したEMQXノード、イベント発生時刻のフィールドを抽出するには、以下のステートメントを使用します。

例：

SELECT
  from_clientid,
  from_username,
  topic,
  qos,
  node,
  timestamp
FROM
  "$events/message_acked"

出力例：

{
  "topic": "t/a",
  "timestamp": 1645002965664,
  "qos": 1,
  "node": "emqx@127.0.0.1",
  "from_username": "u_emqx_1",
  "from_clientid": "c_emqx_1"
}

各フィールドの詳細は以下の通りです。

コード	説明
id	MQTTメッセージID
from_clientid	パブリッシャーのクライアントID
from_username	パブリッシャーのユーザー名
clientid	サブスクライバーのクライアントID
username	サブスクライバーのユーザー名
payload	MQTTペイロード
peerhost	クライアントのIPアドレス
topic	MQTTトピック
qos	QoSレベル
flags	フラグ
pub_props	PUBLISHプロパティ（MQTT 5.0のみ）
puback_props	PUBACKプロパティ（MQTT 5.0のみ）
timestamp	イベント発生時刻（単位：ミリ秒）
publish_received_at	PUBLISHメッセージがEMQXに到達した時刻（単位：ミリ秒）
node	イベントが発生したEMQXノード

ルーティング時メッセージ破棄イベント ("$events/message_dropped")

このイベントトピックは、メッセージがルーティング中に破棄された際にルールをトリガーするために使用できます。

例えば、"$events/message_dropped"イベントトピックから、破棄理由、メッセージトピック、メッセージQoS、イベントが発生したEMQXノード、イベント発生時刻のフィールドを抽出するには、以下のステートメントを使用します。

例：

SELECT
  reason,
  topic,
  qos,
  node,
  timestamp
FROM
  "$events/message_dropped"

出力例：

{
  "topic": "t/a",
  "timestamp": 1645003103004,
  "reason": "no_subscribers",
  "qos": 1,
  "node": "emqx@127.0.0.1"
}

フィールド	説明
id	MQTTメッセージID
reason	破棄理由： no_subscribers: トピックにサブスクライブしているクライアントがいない receive_maximum_exceeded: awaiting_relキューが満杯 packet_identifier_inuse: 未解放のパケットIDを持つQoS 2メッセージを受信した場合
clientid	パブリッシャーのクライアントID
username	パブリッシャーのユーザー名
payload	MQTTペイロード
peerhost	クライアントのIPアドレス
topic	MQTTトピック
qos	QoSレベル
flags	フラグ
pub_props	PUBLISHプロパティ（MQTT 5.0のみ）
timestamp	イベント発生時刻（単位：ミリ秒）
publish_received_at	PUBLISHメッセージがEMQXに到達した時刻（単位：ミリ秒）
node	イベントが発生したノード

配信時メッセージ破棄イベント ("$events/delivery_dropped")

このイベントトピックは、メッセージが配信中に破棄された際にルールをトリガーするために使用できます。

例えば、"$events/delivery_dropped"イベントトピックから、パブリッシャーのIDとユーザー名、破棄理由、メッセージトピックとQoSのフィールドを抽出するには、以下のステートメントを使用します。

例：

SELECT
  from_clientid,
  from_username,
  reason,
  topic,
  qos
FROM "$events/delivery_dropped"

出力例：

{
  "topic": "t/a",
  "reason": "queue_full",
  "qos": 1,
  "from_username": "u_emqx_1",
  "from_clientid": "c_emqx_1"
}

各フィールドの詳細は以下の通りです。

フィールド	説明
id	MQTTメッセージID
reason	破棄理由： queue_full: メッセージ（QoS>0）キューが満杯 no_local: クライアントが自身のパブリッシュしたメッセージを受信できない expired: メッセージまたはセッションが期限切れ qos0_msg: メッセージ（QoS 0）キューが満杯
from_clientid	パブリッシャーのクライアントID
from_username	パブリッシャーのユーザー名
clientid	サブスクライバーのクライアントID
username	サブスクライバーのユーザー名
payload	MQTTペイロード
peerhost	クライアントのIPアドレス
topic	MQTTトピック
qos	メッセージのQoS
flags	フラグ
pub_props	PUBLISHプロパティ（MQTT 5.0クライアントのみ）
timestamp	イベント発生時刻（単位：ミリ秒）
publish_received_at	PUBLISHメッセージがEMQXに到達した時刻（単位：ミリ秒）
node	イベントが発生したEMQXノード

接続完了イベント ("$events/client_connected")

このイベントトピックは、クライアントが正常に接続した際にルールをトリガーするために使用できます。

例えば、"$events/client_connected"イベントトピックから、クライアントID、ユーザー名、キープアライブ間隔、接続したMQTTクライアントがブリッジとして動作しているかどうかのフィールドを抽出するには、以下のステートメントを使用します。

例：

SELECT
  clientid,
  username,
  keepalive,
  is_bridge
FROM
  "$events/client_connected"

出力例：

{
  "username": "u_emqx",
  "keepalive": 60,
  "is_bridge": false,
  "clientid": "c_emqx"
}

受信したMQTTメッセージから選択可能なフィールドは以下の通りです。

フィールド	説明
clientid	クライアントID
username	クライアントのユーザー名
mountpoint	ブリッジメッセージのマウントポイント
peername	クライアントのIPアドレスとポート番号
sockname	EMQXがリッスンしているIPアドレスとポート番号
proto_name	プロトコル名
proto_ver	プロトコルバージョン
keepalive	MQTTキープアライブ間隔
clean_start	MQTTのclean_startフラグ
expiry_interval	MQTTセッションの有効期限
is_bridge	クライアントがブリッジとして動作しているかどうか
connected_at	クライアントの接続完了時刻（単位：ミリ秒）
conn_props	CONNECTプロパティ（MQTT 5.0クライアントのみ）
timestamp	イベント発生時刻（単位：ミリ秒）
node	イベントが発生したEMQXノード
client_attrs	クライアント属性

切断イベント ("$events/client_disconnected")

このイベントトピックは、クライアントが切断された際にルールをトリガーするために使用できます。

例えば、"$events/client_disconnected"イベントトピックから、クライアントID、ユーザー名、切断理由、切断時刻、イベントが発生したEMQXノードのフィールドを抽出するには、以下のステートメントを使用します。

例：

SELECT
  clientid,
  username,
  reason,
  disconnected_at,
  node
FROM
  "$events/client_disconnected"

出力例：

{
  "username": "u_emqx",
  "reason": "normal",
  "node": "emqx@127.0.0.1",
  "disconnected_at": 1645003578536,
  "clientid": "c_emqx"
}

フィールド	説明
reason	切断理由 normal: クライアントが意図的に切断 kicked: REST APIによって強制切断 keepalive_timeout: キープアライブ時間切れ not_authorized: 認可失敗 tcp_closed: ピアがネットワーク接続を閉じた discarded: clean_startがtrueの別クライアントが同じClientIDで接続し、前接続が切断 takenover: clean_startがfalseの別クライアントが同じClientIDで接続し、前接続を引き継ぎ internal_error: 不正なメッセージ形式や未知のエラーによる切断
clientid	クライアントID
username	クライアントのユーザー名
peername	クライアントのIPアドレスとポート番号
sockname	EMQXがリッスンしているIPアドレスとポート番号
disconnected_at	クライアントの切断完了時刻（単位：ミリ秒）
disconn_props	DISCONNECTプロパティ（MQTT 5.0クライアントのみ）
timestamp	イベント発生時刻（単位：ミリ秒）
node	イベントが発生したEMQXノード
client_attrs	クライアント属性

接続応答イベント ("$events/client_connack")

このイベントトピックは、EMQXがクライアントにCONNACKパケットを送信した際にルールをトリガーするために使用できます。

例：

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

出力例：

{
  "username": "u_emqx",
  "reason_code": "success",
  "node": "emqx@127.0.0.1",
  "connected_at": 1645003578536,
  "clientid": "c_emqx"
}

抽出可能なフィールドは以下の通りです。

フィールド	説明
reason_code	理由コード*
clientid	パブリッシャーのクライアントID
username	パブリッシャーのユーザー名
peername	IPアドレスとポート
sockname	EMQXがリッスンしているIPアドレスとポート
proto_name	プロトコル名
proto_ver	プロトコルバージョン
keepalive	MQTTキープアライブ間隔
clean_start	MQTTのclean_startフラグ
expiry_interval	MQTTセッションの有効期限
conn_props	CONNECTプロパティ（MQTT 5.0クライアントのみ）
timestamp	イベント発生時刻（単位：ミリ秒）
node	イベントが発生したEMQXノード

[^*]: MQTT v5.0プロトコルでは、リターンコードが理由コードに名称変更され、より多くのエラータイプを示す理由コードが追加されています（Reason code and ACK - MQTT 5.0 new features）。

以下はMQTT v3.1.1およびMQTT v5.0の理由コード一覧です。

理由コード	説明
connection_accepted	接続が受理された
unacceptable_protocol_version	クライアントが要求したMQTTプロトコルバージョンをEMQXがサポートしていない
client_identifier_not_valid	クライアントIDがEMQXで許可されていない
server_unavaliable	ネットワーク接続は確立されたが、MQTTサービスが利用不可
malformed_username_or_password	ユーザー名またはパスワードのデータ形式が不正
unauthorized_client	クライアント接続が認可されていない

理由コード	説明
success	接続成功
unspecified_error	不明なエラー
malformed_packet	パケットが不正
protocol_error	プロトコルエラー
implementation_specific_error	実装固有のエラー
unsupported_protocol_version	サポートされていないプロトコルバージョン
client_identifier_not_valid	無効なクライアントID
bad_username_or_password	無効なユーザー名またはパスワード
not_authorized	認可されていない
server_unavailable	サーバー利用不可
server_busy	サーバーがビジー状態
banned	接続禁止
bad_authentication_method	無効な認証方式
topic_name_invalid	無効なトピック名
packet_too_large	パケットが大きすぎる
quota_exceeded	クォータ超過
retain_not_supported	Retainメッセージ機能がサポートされていない
qos_not_supported	サポートされていないQoSレベル
use_another_server	別のブローカーを使用してください
server_moved	ブローカーが移動した
connection_rate_exceeded	接続レート制限に達した

認可チェック完了イベント ("$events/client_check_authz_complete")

このイベントトピックは、クライアントの認可チェックが完了した際にルールをトリガーするために使用できます。

例：

SELECT
  clientid,
  username,
  topic,
  action,
  result,
  authz_source,
  node
FROM
  "$events/client_check_authz_complete"

出力例：

{
  "username": "u_emqx",
  "topic": "t/a",
  "action": "publish",
  "result": "allow",
  "authz_source": "cache",
  "node": "emqx@127.0.0.1",
  "clientid": "c_emqx"
}

抽出可能なフィールドは以下の通りです。

フィールド	説明
clientid	クライアントID
username	ユーザー名
peerhost	クライアントのIPアドレス
topic	MQTTトピック
action	パブリッシュまたはサブスクライブのアクション
result	アクセス制御チェックの結果
authz_source	認可のソース
timestamp	タイムスタンプ（単位：ミリ秒）
node	イベントが発生したEMQXノード
client_attrs	クライアント属性

認証チェック完了イベント ("$events/client_check_authn_complete")

このイベントトピックは、クライアントの認証チェックが完了した際にルールをトリガーするために使用できます。

例：

SELECT
  clientid,
  username,
  reason_code,
  is_superuser,
  is_anonymous
FROM
  "$events/client_check_authn_complete"

出力例：

{
  "clientid": "c_emqx",
  "username": "u_emqx",
  "reason_code": "success",
  "is_superuser": true,
  "is_anonymous": false
}

抽出可能なフィールドは以下の通りです。

フィールド	説明
clientid	クライアントID
username	ユーザー名
peername	クライアントのIPアドレス
reason_code	認証結果
is_superuser	このクライアントがスーパーユーザーかどうか
is_anonymous	このクライアントが匿名ユーザーかどうか
client_attrs	クライアント属性

サブスクライブ完了イベント ("$events/session_subscribed")

このイベントトピックは、クライアントが正常にサブスクライブした際にルールをトリガーするために使用できます。

例：

SELECT
  clientid,
  username,
  topic,
  qos
FROM
  "$events/session_subscribed"

出力例：

{
  "username": "u_emqx",
  "topic": "t/a",
  "qos": 1,
  "clientid": "c_emqx"
}

抽出可能なフィールドは以下の通りです。

フィールド	説明
clientid	クライアントID
username	クライアントのユーザー名
peerhost	クライアントのIPアドレス
topic	MQTTトピック
qos	QoSレベル
sub_props	SUBSCRIBEプロパティ（MQTT 5.0クライアントのみ）
timestamp	イベント発生時刻（単位：ミリ秒）
node	イベントが発生したEMQXノード
client_attrs	クライアント属性

サブスクライブ解除イベント ("$events/session_unsubscribed")

このイベントトピックは、クライアントのサブスクリプションが正常に解除された際にルールをトリガーします。

例：

SELECT
  clientid,
  username,
  topic,
  qos
FROM
  "$events/session_unsubscribed"

出力例：

{
  "username": "u_emqx",
  "topic": "t/a",
  "qos": 1,
  "clientid": "c_emqx"
}

抽出可能なフィールドは以下の通りです。

フィールド	説明
clientid	クライアントID
username	クライアントのユーザー名
peerhost	クライアントのIPアドレス
topic	MQTTトピック
qos	QoSレベル
unsub_props	UNSUBSCRIBEプロパティ（MQTT 5.0クライアントのみ）
timestamp	イベント発生時刻（単位：ミリ秒）
node	イベントが発生したEMQXノード
client_attrs	クライアント属性

システムアラーム発動イベント ("$events/sys/alarm_activated")

このイベントトピックは、EMQXのシステムアラームが発動した際にルールをトリガーするために使用できます。

例えば、"$events/sys/alarm_activated"イベントトピックから、アラーム名、詳細、説明メッセージ、発動時刻のフィールドを抽出するには、以下のステートメントを使用します。

例：

SELECT
  name,
  details,
  message,
  activated_at,
  node
FROM
  "$events/sys/alarm_activated"

出力例：

{
  "name": "too_many_processes",
  "details": {
    "usage": "99%",
    "high_watermark": "80%"
  },
  "message": "99% process usage",
  "activated_at": 1645003578536000,
  "node": "emqx@127.0.0.1"
}

抽出可能なフィールドは以下の通りです。

フィールド	説明
name	アラームの短い識別子（例: "too_many_processes"）
details	アラームに関する追加詳細を含むJSONオブジェクト（スキーマは固定されていません。例: {"usage": "99%", "high_watermark": "80%"}）
message	アラームの説明メッセージ（例: "99% process usage"）
activated_at	アラームが発動したUnixタイムスタンプ（マイクロ秒単位）
node	イベントが発生したEMQXノード

システムアラーム解除イベント ("$events/sys/alarm_deactivated")

このイベントトピックは、EMQXのシステムアラームが解除された際にルールをトリガーします。

例えば、"$events/sys/alarm_deactivated"イベントトピックから、アラーム名、詳細、説明メッセージ、発動時刻、解除時刻のフィールドを抽出するには、以下のステートメントを使用します。

例：

SELECT
  name,
  details,
  message,
  activated_at,
  deactivated_at,
  node
FROM
  "$events/sys/alarm_deactivated"

出力例：

{
  "name": "too_many_processes",
  "details": {
    "usage": "99%",
    "high_watermark": "80%"
  },
  "message": "99% process usage",
  "activated_at": 1645003578536000,
  "deactivated_at": 1645004000000000,
  "node": "emqx@127.0.0.1"
}

抽出可能なフィールドは以下の通りです。

フィールド	説明
name	アラームの短い識別子（例: "too_many_processes"）
details	アラームに関する追加詳細を含むJSONオブジェクト（スキーマは固定されていません。例: {"usage": "99%", "high_watermark": "80%"}）
message	アラームの説明メッセージ（例: "99% process usage"）
activated_at	アラームが発動したUnixタイムスタンプ（マイクロ秒単位）
deactivated_at	アラームが解除されたUnixタイムスタンプ（マイクロ秒単位）
node	イベントが発生したEMQXノード

データブリッジ

ルールでは、$bridges/で始まるトピックを使って、データブリッジによってトリガーされたメッセージやイベントを扱います。フォーマットは以下の通りです。

$bridges/<type>:<name>

ここで、

• <type>:<name>はブリッジID、
• <type>はブリッジの種類、
• <name>はブリッジ名です。

例えば、MQTTブリッジのイベントは"$bridges/mqtt:*"の形式で参照できます。MQTTデータブリッジmy_mqtt_bridgeが送信するすべてのメッセージに対してルールを設定するには、以下のステートメントを使用します。

例：

SELECT
  *
FROM
  "$bridges/mqtt:my_mqtt_bridge"

出力例：

{
  "id": "0005E27C1D24E44FF440000017520000",
  "server": "broker.emqx.io:1883",
  "payload": "hello",
  "topic": "t/a",
  "qos": 1,
  "dup": false,
  "retain": false,
  "pub_props": {
    "Message-Expiry-Interval": 30,
    "Payload-Format-Indicator": 0,
    "User-Property": {
      "foo": "bar"
    },
    "User-Property-Pairs": [
      {
        "key": "foo"
      },
      {
        "value": "bar"
      }
    ]
  },
  "message_received_at": 1645002753259,
}

返される各フィールドの説明は以下の通りです。

フィールド	説明
id	MQTTメッセージID
server	リモートMQTTブローカーのサーバー名（例："broker.emqx.io:1883"）
payload	MQTTペイロード
topic	MQTTトピック
qos	MQTTのQoS
dup	MQTTのDUPフラグ
retain	MQTTのRetainフラグ
pub_props	PUBLISHプロパティ（MQTT 5.0クライアントのみ）
message_received_at	メッセージ受信時のタイムスタンプ（単位：ミリ秒）