692 lines
27 KiB
Erlang
692 lines
27 KiB
Erlang
%%--------------------------------------------------------------------
|
|
%% Copyright (c) 2020-2021 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
%%
|
|
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
%% you may not use this file except in compliance with the License.
|
|
%% You may obtain a copy of the License at
|
|
%%
|
|
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
%%
|
|
%% Unless required by applicable law or agreed to in writing, software
|
|
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
%% See the License for the specific language governing permissions and
|
|
%% limitations under the License.
|
|
%%--------------------------------------------------------------------
|
|
|
|
-module(emqx_rule_events).
|
|
|
|
-include("rule_engine.hrl").
|
|
-include_lib("emqx/include/emqx.hrl").
|
|
-include_lib("emqx/include/logger.hrl").
|
|
|
|
-logger_header("[RuleEvents]").
|
|
|
|
-export([ load/1
|
|
, unload/0
|
|
, unload/1
|
|
, event_name/1
|
|
, eventmsg_publish/1
|
|
]).
|
|
|
|
-export([ on_client_connected/3
|
|
, on_client_disconnected/4
|
|
, on_session_subscribed/4
|
|
, on_session_unsubscribed/4
|
|
, on_message_publish/2
|
|
, on_message_dropped/4
|
|
, on_message_delivered/3
|
|
, on_message_acked/3
|
|
, on_delivery_dropped/4
|
|
]).
|
|
|
|
-export([ event_info/0
|
|
, columns/1
|
|
, columns_with_exam/1
|
|
]).
|
|
|
|
-define(SUPPORTED_HOOK,
|
|
[ 'client.connected'
|
|
, 'client.disconnected'
|
|
, 'session.subscribed'
|
|
, 'session.unsubscribed'
|
|
, 'message.publish'
|
|
, 'message.delivered'
|
|
, 'message.acked'
|
|
, 'message.dropped'
|
|
, 'delivery.dropped'
|
|
]).
|
|
|
|
-ifdef(TEST).
|
|
-export([ reason/1
|
|
, hook_fun/1
|
|
, printable_maps/1
|
|
]).
|
|
-endif.
|
|
|
|
load(Topic) ->
|
|
HookPoint = event_name(Topic),
|
|
emqx_hooks:put(HookPoint, {?MODULE, hook_fun(HookPoint),
|
|
[hook_conf(HookPoint, env())]}).
|
|
|
|
unload() ->
|
|
lists:foreach(fun(HookPoint) ->
|
|
emqx_hooks:del(HookPoint, {?MODULE, hook_fun(HookPoint)})
|
|
end, ?SUPPORTED_HOOK).
|
|
|
|
unload(Topic) ->
|
|
HookPoint = event_name(Topic),
|
|
emqx_hooks:del(HookPoint, {?MODULE, hook_fun(HookPoint)}).
|
|
|
|
env() ->
|
|
application:get_all_env(?APP).
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% Callbacks
|
|
%%--------------------------------------------------------------------
|
|
|
|
on_message_publish(Message = #message{flags = #{event := true}},
|
|
_Env) ->
|
|
{ok, Message};
|
|
on_message_publish(Message = #message{flags = #{sys := true}},
|
|
#{ignore_sys_message := true}) ->
|
|
{ok, Message};
|
|
on_message_publish(Message = #message{topic = Topic}, _Env) ->
|
|
case emqx_rule_registry:get_rules_for(Topic) of
|
|
[] -> ok;
|
|
Rules -> emqx_rule_runtime:apply_rules(Rules, eventmsg_publish(Message))
|
|
end,
|
|
{ok, Message}.
|
|
|
|
on_client_connected(ClientInfo, ConnInfo, Env) ->
|
|
may_publish_and_apply('client.connected',
|
|
fun() -> eventmsg_connected(ClientInfo, ConnInfo) end, Env).
|
|
|
|
on_client_disconnected(ClientInfo, Reason, ConnInfo, Env) ->
|
|
may_publish_and_apply('client.disconnected',
|
|
fun() -> eventmsg_disconnected(ClientInfo, ConnInfo, Reason) end, Env).
|
|
|
|
on_session_subscribed(ClientInfo, Topic, SubOpts, Env) ->
|
|
may_publish_and_apply('session.subscribed',
|
|
fun() -> eventmsg_sub_or_unsub('session.subscribed', ClientInfo, Topic, SubOpts) end, Env).
|
|
|
|
on_session_unsubscribed(ClientInfo, Topic, SubOpts, Env) ->
|
|
may_publish_and_apply('session.unsubscribed',
|
|
fun() -> eventmsg_sub_or_unsub('session.unsubscribed', ClientInfo, Topic, SubOpts) end, Env).
|
|
|
|
on_message_dropped(Message = #message{flags = #{sys := true}},
|
|
_, _, #{ignore_sys_message := true}) ->
|
|
{ok, Message};
|
|
on_message_dropped(Message, _, Reason, Env) ->
|
|
may_publish_and_apply('message.dropped',
|
|
fun() -> eventmsg_dropped(Message, Reason) end, Env),
|
|
{ok, Message}.
|
|
|
|
on_message_delivered(_ClientInfo, Message = #message{flags = #{sys := true}},
|
|
#{ignore_sys_message := true}) ->
|
|
{ok, Message};
|
|
on_message_delivered(ClientInfo, Message, Env) ->
|
|
may_publish_and_apply('message.delivered',
|
|
fun() -> eventmsg_delivered(ClientInfo, Message) end, Env),
|
|
{ok, Message}.
|
|
|
|
on_message_acked(_ClientInfo, Message = #message{flags = #{sys := true}},
|
|
#{ignore_sys_message := true}) ->
|
|
{ok, Message};
|
|
on_message_acked(ClientInfo, Message, Env) ->
|
|
may_publish_and_apply('message.acked',
|
|
fun() -> eventmsg_acked(ClientInfo, Message) end, Env),
|
|
{ok, Message}.
|
|
|
|
on_delivery_dropped(_ClientInfo, Message = #message{flags = #{sys := true}},
|
|
_Reason, #{ignore_sys_message := true}) ->
|
|
{ok, Message};
|
|
on_delivery_dropped(ClientInfo, Message, Reason, Env) ->
|
|
may_publish_and_apply('delivery.dropped',
|
|
fun() -> eventmsg_delivery_dropped(ClientInfo, Message, Reason) end, Env),
|
|
{ok, Message}.
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% Event Messages
|
|
%%--------------------------------------------------------------------
|
|
|
|
eventmsg_publish(Message = #message{id = Id, from = ClientId, qos = QoS, flags = Flags, topic = Topic, headers = Headers, payload = Payload, timestamp = Timestamp}) ->
|
|
with_basic_columns('message.publish',
|
|
#{id => emqx_guid:to_hexstr(Id),
|
|
clientid => ClientId,
|
|
username => emqx_message:get_header(username, Message, undefined),
|
|
payload => Payload,
|
|
peerhost => ntoa(emqx_message:get_header(peerhost, Message, undefined)),
|
|
topic => Topic,
|
|
qos => QoS,
|
|
flags => Flags,
|
|
pub_props => printable_maps(emqx_message:get_header(properties, Message, #{})),
|
|
%% the column 'headers' will be removed in the next major release
|
|
headers => printable_maps(Headers),
|
|
publish_received_at => Timestamp
|
|
}).
|
|
|
|
eventmsg_connected(_ClientInfo = #{
|
|
clientid := ClientId,
|
|
username := Username,
|
|
is_bridge := IsBridge,
|
|
mountpoint := Mountpoint
|
|
},
|
|
_ConnInfo = #{
|
|
peername := PeerName,
|
|
sockname := SockName,
|
|
clean_start := CleanStart,
|
|
proto_name := ProtoName,
|
|
proto_ver := ProtoVer,
|
|
keepalive := Keepalive,
|
|
connected_at := ConnectedAt,
|
|
conn_props := ConnProps,
|
|
receive_maximum := RcvMax,
|
|
expiry_interval := ExpiryInterval
|
|
}) ->
|
|
with_basic_columns('client.connected',
|
|
#{clientid => ClientId,
|
|
username => Username,
|
|
mountpoint => Mountpoint,
|
|
peername => ntoa(PeerName),
|
|
sockname => ntoa(SockName),
|
|
proto_name => ProtoName,
|
|
proto_ver => ProtoVer,
|
|
keepalive => Keepalive,
|
|
clean_start => CleanStart,
|
|
receive_maximum => RcvMax,
|
|
expiry_interval => ExpiryInterval,
|
|
is_bridge => IsBridge,
|
|
conn_props => printable_maps(ConnProps),
|
|
connected_at => ConnectedAt
|
|
}).
|
|
|
|
eventmsg_disconnected(_ClientInfo = #{
|
|
clientid := ClientId,
|
|
username := Username
|
|
},
|
|
ConnInfo = #{
|
|
peername := PeerName,
|
|
sockname := SockName,
|
|
disconnected_at := DisconnectedAt
|
|
}, Reason) ->
|
|
with_basic_columns('client.disconnected',
|
|
#{reason => reason(Reason),
|
|
clientid => ClientId,
|
|
username => Username,
|
|
peername => ntoa(PeerName),
|
|
sockname => ntoa(SockName),
|
|
disconn_props => printable_maps(maps:get(disconn_props, ConnInfo, #{})),
|
|
disconnected_at => DisconnectedAt
|
|
}).
|
|
|
|
eventmsg_sub_or_unsub(Event, _ClientInfo = #{
|
|
clientid := ClientId,
|
|
username := Username,
|
|
peerhost := PeerHost
|
|
}, Topic, SubOpts = #{qos := QoS}) ->
|
|
PropKey = sub_unsub_prop_key(Event),
|
|
with_basic_columns(Event,
|
|
#{clientid => ClientId,
|
|
username => Username,
|
|
peerhost => ntoa(PeerHost),
|
|
PropKey => printable_maps(maps:get(PropKey, SubOpts, #{})),
|
|
topic => Topic,
|
|
qos => QoS
|
|
}).
|
|
|
|
eventmsg_dropped(Message = #message{id = Id, from = ClientId, qos = QoS, flags = Flags, topic = Topic, headers = Headers, payload = Payload, timestamp = Timestamp}, Reason) ->
|
|
with_basic_columns('message.dropped',
|
|
#{id => emqx_guid:to_hexstr(Id),
|
|
reason => Reason,
|
|
clientid => ClientId,
|
|
username => emqx_message:get_header(username, Message, undefined),
|
|
payload => Payload,
|
|
peerhost => ntoa(emqx_message:get_header(peerhost, Message, undefined)),
|
|
topic => Topic,
|
|
qos => QoS,
|
|
flags => Flags,
|
|
%% the column 'headers' will be removed in the next major release
|
|
headers => printable_maps(Headers),
|
|
pub_props => printable_maps(emqx_message:get_header(properties, Message, #{})),
|
|
publish_received_at => Timestamp
|
|
}).
|
|
|
|
eventmsg_delivery_dropped(_ClientInfo = #{
|
|
peerhost := PeerHost,
|
|
clientid := ReceiverCId,
|
|
username := ReceiverUsername
|
|
},
|
|
Message = #message{id = Id, from = ClientId, qos = QoS, flags = Flags, topic = Topic,
|
|
headers = Headers, payload = Payload, timestamp = Timestamp},
|
|
Reason) ->
|
|
with_basic_columns('delivery.dropped',
|
|
#{id => emqx_guid:to_hexstr(Id),
|
|
reason => Reason,
|
|
from_clientid => ClientId,
|
|
from_username => emqx_message:get_header(username, Message, undefined),
|
|
clientid => ReceiverCId,
|
|
username => ReceiverUsername,
|
|
payload => Payload,
|
|
peerhost => ntoa(PeerHost),
|
|
topic => Topic,
|
|
qos => QoS,
|
|
flags => Flags,
|
|
%% the column 'headers' will be removed in the next major release
|
|
headers => printable_maps(Headers),
|
|
pub_props => printable_maps(emqx_message:get_header(properties, Message, #{})),
|
|
publish_received_at => Timestamp
|
|
}).
|
|
|
|
eventmsg_delivered(_ClientInfo = #{
|
|
peerhost := PeerHost,
|
|
clientid := ReceiverCId,
|
|
username := ReceiverUsername
|
|
}, Message = #message{id = Id, from = ClientId, qos = QoS, flags = Flags, topic = Topic, headers = Headers, payload = Payload, timestamp = Timestamp}) ->
|
|
with_basic_columns('message.delivered',
|
|
#{id => emqx_guid:to_hexstr(Id),
|
|
from_clientid => ClientId,
|
|
from_username => emqx_message:get_header(username, Message, undefined),
|
|
clientid => ReceiverCId,
|
|
username => ReceiverUsername,
|
|
payload => Payload,
|
|
peerhost => ntoa(PeerHost),
|
|
topic => Topic,
|
|
qos => QoS,
|
|
flags => Flags,
|
|
%% the column 'headers' will be removed in the next major release
|
|
headers => printable_maps(Headers),
|
|
pub_props => printable_maps(emqx_message:get_header(properties, Message, #{})),
|
|
publish_received_at => Timestamp
|
|
}).
|
|
|
|
eventmsg_acked(_ClientInfo = #{
|
|
peerhost := PeerHost,
|
|
clientid := ReceiverCId,
|
|
username := ReceiverUsername
|
|
}, Message = #message{id = Id, from = ClientId, qos = QoS, flags = Flags, topic = Topic, headers = Headers, payload = Payload, timestamp = Timestamp}) ->
|
|
with_basic_columns('message.acked',
|
|
#{id => emqx_guid:to_hexstr(Id),
|
|
from_clientid => ClientId,
|
|
from_username => emqx_message:get_header(username, Message, undefined),
|
|
clientid => ReceiverCId,
|
|
username => ReceiverUsername,
|
|
payload => Payload,
|
|
peerhost => ntoa(PeerHost),
|
|
topic => Topic,
|
|
qos => QoS,
|
|
flags => Flags,
|
|
%% the column 'headers' will be removed in the next major release
|
|
headers => printable_maps(Headers),
|
|
pub_props => printable_maps(emqx_message:get_header(properties, Message, #{})),
|
|
puback_props => printable_maps(emqx_message:get_header(puback_props, Message, #{})),
|
|
publish_received_at => Timestamp
|
|
}).
|
|
|
|
sub_unsub_prop_key('session.subscribed') -> sub_props;
|
|
sub_unsub_prop_key('session.unsubscribed') -> unsub_props.
|
|
|
|
with_basic_columns(EventName, Data) when is_map(Data) ->
|
|
Data#{event => EventName,
|
|
timestamp => erlang:system_time(millisecond),
|
|
node => node()
|
|
}.
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% Events publishing and rules applying
|
|
%%--------------------------------------------------------------------
|
|
|
|
may_publish_and_apply(EventName, GenEventMsg, #{enabled := true, qos := QoS}) ->
|
|
EventTopic = event_topic(EventName),
|
|
EventMsg = GenEventMsg(),
|
|
case emqx_json:safe_encode(EventMsg) of
|
|
{ok, Payload} ->
|
|
_ = emqx_broker:safe_publish(make_msg(QoS, EventTopic, Payload)),
|
|
ok;
|
|
{error, _Reason} ->
|
|
?LOG(error, "Failed to encode event msg for ~p, msg: ~p", [EventName, EventMsg])
|
|
end,
|
|
emqx_rule_runtime:apply_rules(emqx_rule_registry:get_rules_for(EventTopic), EventMsg);
|
|
may_publish_and_apply(EventName, GenEventMsg, _Env) ->
|
|
EventTopic = event_topic(EventName),
|
|
case emqx_rule_registry:get_rules_for(EventTopic) of
|
|
[] -> ok;
|
|
Rules -> emqx_rule_runtime:apply_rules(Rules, GenEventMsg())
|
|
end.
|
|
|
|
make_msg(QoS, Topic, Payload) ->
|
|
emqx_message:set_flags(#{sys => true, event => true},
|
|
emqx_message:make(emqx_events, QoS, Topic, iolist_to_binary(Payload))).
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% Columns
|
|
%%--------------------------------------------------------------------
|
|
columns(Event) ->
|
|
[Key || {Key, _ExampleVal} <- columns_with_exam(Event)].
|
|
|
|
event_info() ->
|
|
[ event_info_message_publish()
|
|
, event_info_message_deliver()
|
|
, event_info_message_acked()
|
|
, event_info_message_dropped()
|
|
, event_info_delivery_dropped()
|
|
, event_info_client_connected()
|
|
, event_info_client_disconnected()
|
|
, event_info_session_subscribed()
|
|
, event_info_session_unsubscribed()
|
|
].
|
|
|
|
event_info_message_publish() ->
|
|
event_info_common(
|
|
'message.publish',
|
|
{<<"message publish">>, <<"消息发布"/utf8>>},
|
|
{<<"message publish">>, <<"消息发布"/utf8>>},
|
|
<<"SELECT payload.msg as msg FROM \"t/#\" WHERE msg = 'hello'">>
|
|
).
|
|
event_info_message_deliver() ->
|
|
event_info_common(
|
|
'message.delivered',
|
|
{<<"message delivered">>, <<"消息已投递"/utf8>>},
|
|
{<<"message delivered">>, <<"消息已投递"/utf8>>},
|
|
<<"SELECT * FROM \"$events/message_delivered\" WHERE topic =~ 't/#'">>
|
|
).
|
|
event_info_message_acked() ->
|
|
event_info_common(
|
|
'message.acked',
|
|
{<<"message acked">>, <<"消息应答"/utf8>>},
|
|
{<<"message acked">>, <<"消息应答"/utf8>>},
|
|
<<"SELECT * FROM \"$events/message_acked\" WHERE topic =~ 't/#'">>
|
|
).
|
|
event_info_message_dropped() ->
|
|
event_info_common(
|
|
'message.dropped',
|
|
{<<"message routing-drop">>, <<"消息转发丢弃"/utf8>>},
|
|
{<<"messages are discarded during forwarding, usually because there are no subscribers">>,
|
|
<<"消息在转发的过程中被丢弃，一般是由于没有订阅者"/utf8>>},
|
|
<<"SELECT * FROM \"$events/message_dropped\" WHERE topic =~ 't/#'">>
|
|
).
|
|
event_info_delivery_dropped() ->
|
|
event_info_common(
|
|
'delivery.dropped',
|
|
{<<"message delivery-drop">>, <<"消息投递丢弃"/utf8>>},
|
|
{<<"messages are discarded during delivery, i.e. because the message queue is full">>,
|
|
<<"消息在投递的过程中被丢弃，比如由于消息队列已满"/utf8>>},
|
|
<<"SELECT * FROM \"$events/delivery_dropped\" WHERE topic =~ 't/#'">>
|
|
).
|
|
event_info_client_connected() ->
|
|
event_info_common(
|
|
'client.connected',
|
|
{<<"client connected">>, <<"连接建立"/utf8>>},
|
|
{<<"client connected">>, <<"连接建立"/utf8>>},
|
|
<<"SELECT * FROM \"$events/client_connected\"">>
|
|
).
|
|
event_info_client_disconnected() ->
|
|
event_info_common(
|
|
'client.disconnected',
|
|
{<<"client disconnected">>, <<"连接断开"/utf8>>},
|
|
{<<"client disconnected">>, <<"连接断开"/utf8>>},
|
|
<<"SELECT * FROM \"$events/client_disconnected\" WHERE topic =~ 't/#'">>
|
|
).
|
|
event_info_session_subscribed() ->
|
|
event_info_common(
|
|
'session.subscribed',
|
|
{<<"session subscribed">>, <<"会话订阅完成"/utf8>>},
|
|
{<<"session subscribed">>, <<"会话订阅完成"/utf8>>},
|
|
<<"SELECT * FROM \"$events/session_subscribed\" WHERE topic =~ 't/#'">>
|
|
).
|
|
event_info_session_unsubscribed() ->
|
|
event_info_common(
|
|
'session.unsubscribed',
|
|
{<<"session unsubscribed">>, <<"会话取消订阅完成"/utf8>>},
|
|
{<<"session unsubscribed">>, <<"会话取消订阅完成"/utf8>>},
|
|
<<"SELECT * FROM \"$events/session_unsubscribed\" WHERE topic =~ 't/#'">>
|
|
).
|
|
|
|
event_info_common(Event, {TitleEN, TitleZH}, {DescrEN, DescrZH}, SqlExam) ->
|
|
#{event => event_topic(Event),
|
|
title => #{en => TitleEN, zh => TitleZH},
|
|
description => #{en => DescrEN, zh => DescrZH},
|
|
test_columns => test_columns(Event),
|
|
columns => columns(Event),
|
|
sql_example => SqlExam
|
|
}.
|
|
|
|
test_columns('message.dropped') ->
|
|
[ {<<"reason">>, <<"no_subscribers">>}
|
|
] ++ test_columns('message.publish');
|
|
test_columns('message.publish') ->
|
|
[ {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
, {<<"payload">>, <<"{\"msg\": \"hello\"}">>}
|
|
];
|
|
test_columns('message.acked') ->
|
|
test_columns('message.delivered');
|
|
test_columns('message.delivered') ->
|
|
[ {<<"from_clientid">>, <<"c_emqx_1">>}
|
|
, {<<"from_username">>, <<"u_emqx_1">>}
|
|
, {<<"clientid">>, <<"c_emqx_2">>}
|
|
, {<<"username">>, <<"u_emqx_2">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
, {<<"payload">>, <<"{\"msg\": \"hello\"}">>}
|
|
];
|
|
test_columns('delivery.dropped') ->
|
|
[ {<<"reason">>, <<"queue_full">>}
|
|
] ++ test_columns('message.delivered');
|
|
test_columns('client.connected') ->
|
|
[ {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"peername">>, <<"127.0.0.1:52918">>}
|
|
];
|
|
test_columns('client.disconnected') ->
|
|
[ {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"reason">>, <<"normal">>}
|
|
];
|
|
test_columns('session.unsubscribed') ->
|
|
test_columns('session.subscribed');
|
|
test_columns('session.subscribed') ->
|
|
[ {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
].
|
|
|
|
columns_with_exam('message.publish') ->
|
|
[ {<<"event">>, 'message.publish'}
|
|
, {<<"id">>, emqx_guid:to_hexstr(emqx_guid:gen())}
|
|
, {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"payload">>, <<"{\"msg\": \"hello\"}">>}
|
|
, {<<"peerhost">>, <<"192.168.0.10">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
, {<<"flags">>, #{}}
|
|
, {<<"headers">>, undefined}
|
|
, {<<"publish_received_at">>, erlang:system_time(millisecond)}
|
|
, {<<"timestamp">>, erlang:system_time(millisecond)}
|
|
, {<<"node">>, node()}
|
|
];
|
|
columns_with_exam('message.delivered') ->
|
|
[ {<<"event">>, 'message.delivered'}
|
|
, {<<"id">>, emqx_guid:to_hexstr(emqx_guid:gen())}
|
|
, {<<"from_clientid">>, <<"c_emqx_1">>}
|
|
, {<<"from_username">>, <<"u_emqx_1">>}
|
|
, {<<"clientid">>, <<"c_emqx_2">>}
|
|
, {<<"username">>, <<"u_emqx_2">>}
|
|
, {<<"payload">>, <<"{\"msg\": \"hello\"}">>}
|
|
, {<<"peerhost">>, <<"192.168.0.10">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
, {<<"flags">>, #{}}
|
|
, {<<"publish_received_at">>, erlang:system_time(millisecond)}
|
|
, {<<"timestamp">>, erlang:system_time(millisecond)}
|
|
, {<<"node">>, node()}
|
|
];
|
|
columns_with_exam('message.acked') ->
|
|
[ {<<"event">>, 'message.acked'}
|
|
, {<<"id">>, emqx_guid:to_hexstr(emqx_guid:gen())}
|
|
, {<<"from_clientid">>, <<"c_emqx_1">>}
|
|
, {<<"from_username">>, <<"u_emqx_1">>}
|
|
, {<<"clientid">>, <<"c_emqx_2">>}
|
|
, {<<"username">>, <<"u_emqx_2">>}
|
|
, {<<"payload">>, <<"{\"msg\": \"hello\"}">>}
|
|
, {<<"peerhost">>, <<"192.168.0.10">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
, {<<"flags">>, #{}}
|
|
, {<<"publish_received_at">>, erlang:system_time(millisecond)}
|
|
, {<<"timestamp">>, erlang:system_time(millisecond)}
|
|
, {<<"node">>, node()}
|
|
];
|
|
columns_with_exam('message.dropped') ->
|
|
[ {<<"event">>, 'message.dropped'}
|
|
, {<<"id">>, emqx_guid:to_hexstr(emqx_guid:gen())}
|
|
, {<<"reason">>, no_subscribers}
|
|
, {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"payload">>, <<"{\"msg\": \"hello\"}">>}
|
|
, {<<"peerhost">>, <<"192.168.0.10">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
, {<<"flags">>, #{}}
|
|
, {<<"publish_received_at">>, erlang:system_time(millisecond)}
|
|
, {<<"timestamp">>, erlang:system_time(millisecond)}
|
|
, {<<"node">>, node()}
|
|
];
|
|
columns_with_exam('delivery.dropped') ->
|
|
[ {<<"event">>, 'delivery.dropped'}
|
|
, {<<"id">>, emqx_guid:to_hexstr(emqx_guid:gen())}
|
|
, {<<"reason">>, queue_full}
|
|
, {<<"from_clientid">>, <<"c_emqx_1">>}
|
|
, {<<"from_username">>, <<"u_emqx_1">>}
|
|
, {<<"clientid">>, <<"c_emqx_2">>}
|
|
, {<<"username">>, <<"u_emqx_2">>}
|
|
, {<<"payload">>, <<"{\"msg\": \"hello\"}">>}
|
|
, {<<"peerhost">>, <<"192.168.0.10">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
, {<<"flags">>, #{}}
|
|
, {<<"publish_received_at">>, erlang:system_time(millisecond)}
|
|
, {<<"timestamp">>, erlang:system_time(millisecond)}
|
|
, {<<"node">>, node()}
|
|
];
|
|
columns_with_exam('client.connected') ->
|
|
[ {<<"event">>, 'client.connected'}
|
|
, {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"mountpoint">>, undefined}
|
|
, {<<"peername">>, <<"192.168.0.10:56431">>}
|
|
, {<<"sockname">>, <<"0.0.0.0:1883">>}
|
|
, {<<"proto_name">>, <<"MQTT">>}
|
|
, {<<"proto_ver">>, 5}
|
|
, {<<"keepalive">>, 60}
|
|
, {<<"clean_start">>, true}
|
|
, {<<"expiry_interval">>, 3600}
|
|
, {<<"is_bridge">>, false}
|
|
, {<<"connected_at">>, erlang:system_time(millisecond)}
|
|
, {<<"timestamp">>, erlang:system_time(millisecond)}
|
|
, {<<"node">>, node()}
|
|
];
|
|
columns_with_exam('client.disconnected') ->
|
|
[ {<<"event">>, 'client.disconnected'}
|
|
, {<<"reason">>, normal}
|
|
, {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"peername">>, <<"192.168.0.10:56431">>}
|
|
, {<<"sockname">>, <<"0.0.0.0:1883">>}
|
|
, {<<"disconnected_at">>, erlang:system_time(millisecond)}
|
|
, {<<"timestamp">>, erlang:system_time(millisecond)}
|
|
, {<<"node">>, node()}
|
|
];
|
|
columns_with_exam('session.subscribed') ->
|
|
[ {<<"event">>, 'session.subscribed'}
|
|
, {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"peerhost">>, <<"192.168.0.10">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
, {<<"timestamp">>, erlang:system_time(millisecond)}
|
|
, {<<"node">>, node()}
|
|
];
|
|
columns_with_exam('session.unsubscribed') ->
|
|
[ {<<"event">>, 'session.unsubscribed'}
|
|
, {<<"clientid">>, <<"c_emqx">>}
|
|
, {<<"username">>, <<"u_emqx">>}
|
|
, {<<"peerhost">>, <<"192.168.0.10">>}
|
|
, {<<"topic">>, <<"t/a">>}
|
|
, {<<"qos">>, 1}
|
|
, {<<"timestamp">>, erlang:system_time(millisecond)}
|
|
, {<<"node">>, node()}
|
|
].
|
|
|
|
%%--------------------------------------------------------------------
|
|
%% Helper functions
|
|
%%--------------------------------------------------------------------
|
|
|
|
hook_conf(HookPoint, Env) ->
|
|
Events = proplists:get_value(events, Env, []),
|
|
IgnoreSys = proplists:get_value(ignore_sys_message, Env, true),
|
|
case lists:keyfind(HookPoint, 1, Events) of
|
|
{_, on, QoS} -> #{enabled => true, qos => QoS, ignore_sys_message => IgnoreSys};
|
|
_ -> #{enabled => false, qos => 1, ignore_sys_message => IgnoreSys}
|
|
end.
|
|
|
|
hook_fun(Event) ->
|
|
case string:split(atom_to_list(Event), ".") of
|
|
[Prefix, Name] ->
|
|
list_to_atom(lists:append(["on_", Prefix, "_", Name]));
|
|
[_] ->
|
|
error(invalid_event, Event)
|
|
end.
|
|
|
|
reason(Reason) when is_atom(Reason) -> Reason;
|
|
reason({shutdown, Reason}) when is_atom(Reason) -> Reason;
|
|
reason({Error, _}) when is_atom(Error) -> Error;
|
|
reason(_) -> internal_error.
|
|
|
|
ntoa(undefined) -> undefined;
|
|
ntoa({IpAddr, Port}) ->
|
|
iolist_to_binary([inet:ntoa(IpAddr), ":", integer_to_list(Port)]);
|
|
ntoa(IpAddr) ->
|
|
iolist_to_binary(inet:ntoa(IpAddr)).
|
|
|
|
event_name(<<"$events/client_connected", _/binary>>) -> 'client.connected';
|
|
event_name(<<"$events/client_disconnected", _/binary>>) -> 'client.disconnected';
|
|
event_name(<<"$events/session_subscribed", _/binary>>) -> 'session.subscribed';
|
|
event_name(<<"$events/session_unsubscribed", _/binary>>) ->
|
|
'session.unsubscribed';
|
|
event_name(<<"$events/message_delivered", _/binary>>) -> 'message.delivered';
|
|
event_name(<<"$events/message_acked", _/binary>>) -> 'message.acked';
|
|
event_name(<<"$events/message_dropped", _/binary>>) -> 'message.dropped';
|
|
event_name(<<"$events/delivery_dropped", _/binary>>) -> 'delivery.dropped';
|
|
event_name(_) -> 'message.publish'.
|
|
|
|
event_topic('client.connected') -> <<"$events/client_connected">>;
|
|
event_topic('client.disconnected') -> <<"$events/client_disconnected">>;
|
|
event_topic('session.subscribed') -> <<"$events/session_subscribed">>;
|
|
event_topic('session.unsubscribed') -> <<"$events/session_unsubscribed">>;
|
|
event_topic('message.delivered') -> <<"$events/message_delivered">>;
|
|
event_topic('message.acked') -> <<"$events/message_acked">>;
|
|
event_topic('message.dropped') -> <<"$events/message_dropped">>;
|
|
event_topic('delivery.dropped') -> <<"$events/delivery_dropped">>;
|
|
event_topic('message.publish') -> <<"$events/message_publish">>.
|
|
|
|
printable_maps(undefined) -> #{};
|
|
printable_maps(Headers) ->
|
|
maps:fold(
|
|
fun (K, V0, AccIn) when K =:= peerhost; K =:= peername; K =:= sockname ->
|
|
AccIn#{K => ntoa(V0)};
|
|
('User-Property', V0, AccIn) when is_list(V0) ->
|
|
AccIn#{
|
|
'User-Property' => maps:from_list(V0),
|
|
'User-Property-Pairs' => [#{
|
|
key => Key,
|
|
value => Value
|
|
} || {Key, Value} <- V0]
|
|
};
|
|
(K, V0, AccIn) -> AccIn#{K => V0}
|
|
end, #{}, Headers).
|