Rename the 'client_id' field to 'clientid'
This commit is contained in:
parent
1a5c10bd6f
commit
20ddd498fc
|
@ -28,7 +28,7 @@
|
|||
%% APIs
|
||||
%%--------------------------------------------------------------------
|
||||
|
||||
-spec(authenticate(emqx_types:client_info())
|
||||
-spec(authenticate(emqx_types:clientinfo())
|
||||
-> {ok, #{auth_result := emqx_types:auth_result(),
|
||||
anonymous := boolean}} | {error, term()}).
|
||||
authenticate(Client) ->
|
||||
|
|
|
@ -86,7 +86,7 @@ bin(B) when is_binary(B) ->
|
|||
B.
|
||||
|
||||
%% @doc Match access rule
|
||||
-spec(match(emqx_types:client_info(), emqx_types:topic(), rule())
|
||||
-spec(match(emqx_types:clientinfo(), emqx_types:topic(), rule())
|
||||
-> {matched, allow} | {matched, deny} | nomatch).
|
||||
match(_ClientInfo, _Topic, {AllowDeny, all}) when ?ALLOW_DENY(AllowDeny) ->
|
||||
{matched, AllowDeny};
|
||||
|
@ -104,7 +104,7 @@ match_who(_ClientInfo, {user, all}) ->
|
|||
true;
|
||||
match_who(_ClientInfo, {client, all}) ->
|
||||
true;
|
||||
match_who(#{client_id := ClientId}, {client, ClientId}) ->
|
||||
match_who(#{clientid := ClientId}, {client, ClientId}) ->
|
||||
true;
|
||||
match_who(#{username := Username}, {user, Username}) ->
|
||||
true;
|
||||
|
@ -142,9 +142,9 @@ feed_var(ClientInfo, Pattern) ->
|
|||
feed_var(ClientInfo, Pattern, []).
|
||||
feed_var(_ClientInfo, [], Acc) ->
|
||||
lists:reverse(Acc);
|
||||
feed_var(ClientInfo = #{client_id := undefined}, [<<"%c">>|Words], Acc) ->
|
||||
feed_var(ClientInfo = #{clientid := undefined}, [<<"%c">>|Words], Acc) ->
|
||||
feed_var(ClientInfo, Words, [<<"%c">>|Acc]);
|
||||
feed_var(ClientInfo = #{client_id := ClientId}, [<<"%c">>|Words], Acc) ->
|
||||
feed_var(ClientInfo = #{clientid := ClientId}, [<<"%c">>|Words], Acc) ->
|
||||
feed_var(ClientInfo, Words, [ClientId |Acc]);
|
||||
feed_var(ClientInfo = #{username := undefined}, [<<"%u">>|Words], Acc) ->
|
||||
feed_var(ClientInfo, Words, [<<"%u">>|Acc]);
|
||||
|
|
|
@ -73,11 +73,11 @@ start_link() ->
|
|||
-spec(stop() -> ok).
|
||||
stop() -> gen_server:stop(?MODULE).
|
||||
|
||||
-spec(check(emqx_types:client_info()) -> boolean()).
|
||||
check(#{client_id := ClientId,
|
||||
-spec(check(emqx_types:clientinfo()) -> boolean()).
|
||||
check(#{clientid := ClientId,
|
||||
username := Username,
|
||||
peerhost := IPAddr}) ->
|
||||
ets:member(?BANNED_TAB, {client_id, ClientId})
|
||||
ets:member(?BANNED_TAB, {clientid, ClientId})
|
||||
orelse ets:member(?BANNED_TAB, {username, Username})
|
||||
orelse ets:member(?BANNED_TAB, {ipaddr, IPAddr}).
|
||||
|
||||
|
@ -85,7 +85,7 @@ check(#{client_id := ClientId,
|
|||
add(Banned) when is_record(Banned, banned) ->
|
||||
mnesia:dirty_write(?BANNED_TAB, Banned).
|
||||
|
||||
-spec(delete({client_id, emqx_types:client_id()}
|
||||
-spec(delete({clientid, emqx_types:clientid()}
|
||||
| {username, emqx_types:username()}
|
||||
| {peerhost, emqx_types:peerhost()}) -> ok).
|
||||
delete(Key) -> mnesia:dirty_delete(?BANNED_TAB, Key).
|
||||
|
|
|
@ -94,19 +94,19 @@ start_link() ->
|
|||
|
||||
%% @doc Register a channel.
|
||||
%% Channel will be unregistered automatically when the channel process dies
|
||||
-spec(register_channel(emqx_types:client_id()) -> ok).
|
||||
-spec(register_channel(emqx_types:clientid()) -> ok).
|
||||
register_channel(ClientId) when is_binary(ClientId) ->
|
||||
register_channel(ClientId, self()).
|
||||
|
||||
%% @doc Register a channel with pid.
|
||||
-spec(register_channel(emqx_types:client_id(), chan_pid()) -> ok).
|
||||
-spec(register_channel(emqx_types:clientid(), chan_pid()) -> ok).
|
||||
register_channel(ClientId, ChanPid) ->
|
||||
Chan = {ClientId, ChanPid},
|
||||
true = ets:insert(?CHAN_TAB, Chan),
|
||||
ok = emqx_cm_registry:register_channel(Chan),
|
||||
cast({registered, Chan}).
|
||||
|
||||
-spec(unregister_channel(emqx_types:client_id()) -> ok).
|
||||
-spec(unregister_channel(emqx_types:clientid()) -> ok).
|
||||
unregister_channel(ClientId) when is_binary(ClientId) ->
|
||||
true = do_unregister_channel({ClientId, self()}),
|
||||
ok.
|
||||
|
@ -119,31 +119,31 @@ do_unregister_channel(Chan) ->
|
|||
true = ets:delete(?CHAN_STATS_TAB, Chan),
|
||||
ets:delete_object(?CHAN_TAB, Chan).
|
||||
|
||||
%% @doc Get attrs of a channel.
|
||||
-spec(get_chan_attrs(emqx_types:client_id()) -> maybe(emqx_types:attrs())).
|
||||
%% @doc Get info of a channel.
|
||||
-spec(get_chan_attrs(emqx_types:clientid()) -> maybe(emqx_types:attrs())).
|
||||
get_chan_attrs(ClientId) ->
|
||||
with_channel(ClientId, fun(ChanPid) -> get_chan_attrs(ClientId, ChanPid) end).
|
||||
|
||||
-spec(get_chan_attrs(emqx_types:client_id(), chan_pid()) -> maybe(emqx_types:attrs())).
|
||||
-spec(get_chan_attrs(emqx_types:clientid(), chan_pid()) -> maybe(emqx_types:attrs())).
|
||||
get_chan_attrs(ClientId, ChanPid) when node(ChanPid) == node() ->
|
||||
Chan = {ClientId, ChanPid},
|
||||
emqx_tables:lookup_value(?CHAN_ATTRS_TAB, Chan);
|
||||
get_chan_attrs(ClientId, ChanPid) ->
|
||||
rpc_call(node(ChanPid), get_chan_attrs, [ClientId, ChanPid]).
|
||||
|
||||
%% @doc Set attrs of a channel.
|
||||
-spec(set_chan_attrs(emqx_types:client_id(), emqx_types:attrs()) -> ok).
|
||||
set_chan_attrs(ClientId, Attrs) when is_binary(ClientId) ->
|
||||
%% @doc Set info of a channel.
|
||||
-spec(set_chan_attrs(emqx_types:clientid(), emqx_types:attrs()) -> ok).
|
||||
set_chan_attrs(ClientId, Info) when is_binary(ClientId) ->
|
||||
Chan = {ClientId, self()},
|
||||
true = ets:insert(?CHAN_ATTRS_TAB, {Chan, Attrs}),
|
||||
true = ets:insert(?CHAN_ATTRS_TAB, {Chan, Info}),
|
||||
ok.
|
||||
|
||||
%% @doc Get channel's stats.
|
||||
-spec(get_chan_stats(emqx_types:client_id()) -> maybe(emqx_types:stats())).
|
||||
-spec(get_chan_stats(emqx_types:clientid()) -> maybe(emqx_types:stats())).
|
||||
get_chan_stats(ClientId) ->
|
||||
with_channel(ClientId, fun(ChanPid) -> get_chan_stats(ClientId, ChanPid) end).
|
||||
|
||||
-spec(get_chan_stats(emqx_types:client_id(), chan_pid()) -> maybe(emqx_types:stats())).
|
||||
-spec(get_chan_stats(emqx_types:clientid(), chan_pid()) -> maybe(emqx_types:stats())).
|
||||
get_chan_stats(ClientId, ChanPid) when node(ChanPid) == node() ->
|
||||
Chan = {ClientId, ChanPid},
|
||||
emqx_tables:lookup_value(?CHAN_STATS_TAB, Chan);
|
||||
|
@ -151,23 +151,23 @@ get_chan_stats(ClientId, ChanPid) ->
|
|||
rpc_call(node(ChanPid), get_chan_stats, [ClientId, ChanPid]).
|
||||
|
||||
%% @doc Set channel's stats.
|
||||
-spec(set_chan_stats(emqx_types:client_id(), emqx_types:stats()) -> ok).
|
||||
-spec(set_chan_stats(emqx_types:clientid(), emqx_types:stats()) -> ok).
|
||||
set_chan_stats(ClientId, Stats) when is_binary(ClientId) ->
|
||||
set_chan_stats(ClientId, self(), Stats).
|
||||
|
||||
-spec(set_chan_stats(emqx_types:client_id(), chan_pid(), emqx_types:stats()) -> ok).
|
||||
-spec(set_chan_stats(emqx_types:clientid(), chan_pid(), emqx_types:stats()) -> ok).
|
||||
set_chan_stats(ClientId, ChanPid, Stats) ->
|
||||
Chan = {ClientId, ChanPid},
|
||||
true = ets:insert(?CHAN_STATS_TAB, {Chan, Stats}),
|
||||
ok.
|
||||
|
||||
%% @doc Open a session.
|
||||
-spec(open_session(boolean(), emqx_types:client_info(), emqx_types:conninfo())
|
||||
-spec(open_session(boolean(), emqx_types:clientinfo(), emqx_types:conninfo())
|
||||
-> {ok, #{session := emqx_session:session(),
|
||||
present := boolean(),
|
||||
pendings => list()}}
|
||||
| {error, Reason :: term()}).
|
||||
open_session(true, ClientInfo = #{client_id := ClientId}, ConnInfo) ->
|
||||
open_session(true, ClientInfo = #{clientid := ClientId}, ConnInfo) ->
|
||||
CleanStart = fun(_) ->
|
||||
ok = discard_session(ClientId),
|
||||
Session = emqx_session:init(ClientInfo, ConnInfo),
|
||||
|
@ -175,7 +175,7 @@ open_session(true, ClientInfo = #{client_id := ClientId}, ConnInfo) ->
|
|||
end,
|
||||
emqx_cm_locker:trans(ClientId, CleanStart);
|
||||
|
||||
open_session(false, ClientInfo = #{client_id := ClientId}, ConnInfo) ->
|
||||
open_session(false, ClientInfo = #{clientid := ClientId}, ConnInfo) ->
|
||||
ResumeStart = fun(_) ->
|
||||
case takeover_session(ClientId) of
|
||||
{ok, ConnMod, ChanPid, Session} ->
|
||||
|
@ -192,7 +192,7 @@ open_session(false, ClientInfo = #{client_id := ClientId}, ConnInfo) ->
|
|||
emqx_cm_locker:trans(ClientId, ResumeStart).
|
||||
|
||||
%% @doc Try to takeover a session.
|
||||
-spec(takeover_session(emqx_types:client_id())
|
||||
-spec(takeover_session(emqx_types:clientid())
|
||||
-> {ok, emqx_session:session()} | {error, Reason :: term()}).
|
||||
takeover_session(ClientId) ->
|
||||
case lookup_channels(ClientId) of
|
||||
|
@ -221,7 +221,7 @@ takeover_session(ClientId, ChanPid) ->
|
|||
rpc_call(node(ChanPid), takeover_session, [ClientId, ChanPid]).
|
||||
|
||||
%% @doc Discard all the sessions identified by the ClientId.
|
||||
-spec(discard_session(emqx_types:client_id()) -> ok).
|
||||
-spec(discard_session(emqx_types:clientid()) -> ok).
|
||||
discard_session(ClientId) when is_binary(ClientId) ->
|
||||
case lookup_channels(ClientId) of
|
||||
[] -> ok;
|
||||
|
@ -259,12 +259,12 @@ with_channel(ClientId, Fun) ->
|
|||
end.
|
||||
|
||||
%% @doc Lookup channels.
|
||||
-spec(lookup_channels(emqx_types:client_id()) -> list(chan_pid())).
|
||||
-spec(lookup_channels(emqx_types:clientid()) -> list(chan_pid())).
|
||||
lookup_channels(ClientId) ->
|
||||
lookup_channels(global, ClientId).
|
||||
|
||||
%% @doc Lookup local or global channels.
|
||||
-spec(lookup_channels(local | global, emqx_types:client_id()) -> list(chan_pid())).
|
||||
-spec(lookup_channels(local | global, emqx_types:clientid()) -> list(chan_pid())).
|
||||
lookup_channels(global, ClientId) ->
|
||||
case emqx_cm_registry:is_enabled() of
|
||||
true ->
|
||||
|
|
|
@ -32,11 +32,11 @@
|
|||
start_link() ->
|
||||
ekka_locker:start_link(?MODULE).
|
||||
|
||||
-spec(trans(emqx_types:client_id(), fun(([node()]) -> any())) -> any()).
|
||||
-spec(trans(emqx_types:clientid(), fun(([node()]) -> any())) -> any()).
|
||||
trans(ClientId, Fun) ->
|
||||
trans(ClientId, Fun, undefined).
|
||||
|
||||
-spec(trans(maybe(emqx_types:client_id()),
|
||||
-spec(trans(maybe(emqx_types:clientid()),
|
||||
fun(([node()])-> any()), ekka_locker:piggyback()) -> any()).
|
||||
trans(undefined, Fun, _Piggyback) ->
|
||||
Fun([]);
|
||||
|
@ -48,15 +48,15 @@ trans(ClientId, Fun, Piggyback) ->
|
|||
{error, client_id_unavailable}
|
||||
end.
|
||||
|
||||
-spec(lock(emqx_types:client_id()) -> ekka_locker:lock_result()).
|
||||
-spec(lock(emqx_types:clientid()) -> ekka_locker:lock_result()).
|
||||
lock(ClientId) ->
|
||||
ekka_locker:acquire(?MODULE, ClientId, strategy()).
|
||||
|
||||
-spec(lock(emqx_types:client_id(), ekka_locker:piggyback()) -> ekka_locker:lock_result()).
|
||||
-spec(lock(emqx_types:clientid(), ekka_locker:piggyback()) -> ekka_locker:lock_result()).
|
||||
lock(ClientId, Piggyback) ->
|
||||
ekka_locker:acquire(?MODULE, ClientId, strategy(), Piggyback).
|
||||
|
||||
-spec(unlock(emqx_types:client_id()) -> {boolean(), [node()]}).
|
||||
-spec(unlock(emqx_types:clientid()) -> {boolean(), [node()]}).
|
||||
unlock(ClientId) ->
|
||||
ekka_locker:release(?MODULE, ClientId, strategy()).
|
||||
|
||||
|
|
|
@ -65,8 +65,8 @@ is_enabled() ->
|
|||
emqx:get_env(enable_channel_registry, true).
|
||||
|
||||
%% @doc Register a global channel.
|
||||
-spec(register_channel(emqx_types:client_id()
|
||||
| {emqx_types:client_id(), pid()}) -> ok).
|
||||
-spec(register_channel(emqx_types:clientid()
|
||||
| {emqx_types:clientid(), pid()}) -> ok).
|
||||
register_channel(ClientId) when is_binary(ClientId) ->
|
||||
register_channel({ClientId, self()});
|
||||
|
||||
|
@ -77,8 +77,8 @@ register_channel({ClientId, ChanPid}) when is_binary(ClientId), is_pid(ChanPid)
|
|||
end.
|
||||
|
||||
%% @doc Unregister a global channel.
|
||||
-spec(unregister_channel(emqx_types:client_id()
|
||||
| {emqx_types:client_id(), pid()}) -> ok).
|
||||
-spec(unregister_channel(emqx_types:clientid()
|
||||
| {emqx_types:clientid(), pid()}) -> ok).
|
||||
unregister_channel(ClientId) when is_binary(ClientId) ->
|
||||
unregister_channel({ClientId, self()});
|
||||
|
||||
|
@ -89,7 +89,7 @@ unregister_channel({ClientId, ChanPid}) when is_binary(ClientId), is_pid(ChanPid
|
|||
end.
|
||||
|
||||
%% @doc Lookup the global channels.
|
||||
-spec(lookup_channels(emqx_types:client_id()) -> list(pid())).
|
||||
-spec(lookup_channels(emqx_types:clientid()) -> list(pid())).
|
||||
lookup_channels(ClientId) ->
|
||||
[ChanPid || #channel{pid = ChanPid} <- mnesia:dirty_read(?TAB, ClientId)].
|
||||
|
||||
|
|
|
@ -51,7 +51,7 @@
|
|||
}).
|
||||
|
||||
-record(flapping, {
|
||||
client_id :: emqx_types:client_id(),
|
||||
clientid :: emqx_types:clientid(),
|
||||
peerhost :: emqx_types:peerhost(),
|
||||
started_at :: pos_integer(),
|
||||
detect_cnt :: pos_integer(),
|
||||
|
@ -69,8 +69,8 @@ start_link() ->
|
|||
stop() -> gen_server:stop(?MODULE).
|
||||
|
||||
%% @doc Check flapping when a MQTT client connected.
|
||||
-spec(check(emqx_types:client_info()) -> boolean()).
|
||||
check(#{client_id := ClientId}) ->
|
||||
-spec(check(emqx_types:clientinfo()) -> boolean()).
|
||||
check(#{clientid := ClientId}) ->
|
||||
check(ClientId, get_policy()).
|
||||
|
||||
check(ClientId, #{banned_interval := Interval}) ->
|
||||
|
@ -81,10 +81,10 @@ check(ClientId, #{banned_interval := Interval}) ->
|
|||
end.
|
||||
|
||||
%% @doc Detect flapping when a MQTT client disconnected.
|
||||
-spec(detect(emqx_types:client_info()) -> boolean()).
|
||||
-spec(detect(emqx_types:clientinfo()) -> boolean()).
|
||||
detect(Client) -> detect(Client, get_policy()).
|
||||
|
||||
detect(#{client_id := ClientId, peerhost := PeerHost},
|
||||
detect(#{clientid := ClientId, peerhost := PeerHost},
|
||||
Policy = #{threshold := Threshold}) ->
|
||||
try ets:update_counter(?FLAPPING_TAB, ClientId, {#flapping.detect_cnt, 1}) of
|
||||
Cnt when Cnt < Threshold -> false;
|
||||
|
@ -97,7 +97,7 @@ detect(#{client_id := ClientId, peerhost := PeerHost},
|
|||
catch
|
||||
error:badarg ->
|
||||
%% Create a flapping record.
|
||||
Flapping = #flapping{client_id = ClientId,
|
||||
Flapping = #flapping{clientid = ClientId,
|
||||
peerhost = PeerHost,
|
||||
started_at = emqx_time:now_ms(),
|
||||
detect_cnt = 1
|
||||
|
@ -131,7 +131,7 @@ handle_call(Req, _From, State) ->
|
|||
?LOG(error, "Unexpected call: ~p", [Req]),
|
||||
{reply, ignored, State}.
|
||||
|
||||
handle_cast({detected, Flapping = #flapping{client_id = ClientId,
|
||||
handle_cast({detected, Flapping = #flapping{clientid = ClientId,
|
||||
peerhost = PeerHost,
|
||||
started_at = StartedAt,
|
||||
detect_cnt = DetectCnt},
|
||||
|
@ -142,7 +142,7 @@ handle_cast({detected, Flapping = #flapping{client_id = ClientId,
|
|||
?LOG(error, "Flapping detected: ~s(~s) disconnected ~w times in ~wms",
|
||||
[ClientId, esockd_net:ntoa(PeerHost), DetectCnt, Duration]),
|
||||
%% Banned.
|
||||
BannedFlapping = Flapping#flapping{client_id = {banned, ClientId},
|
||||
BannedFlapping = Flapping#flapping{clientid = {banned, ClientId},
|
||||
banned_at = emqx_time:now_ms()
|
||||
},
|
||||
alarm_handler:set_alarm({{flapping_detected, ClientId}, BannedFlapping}),
|
||||
|
@ -192,11 +192,11 @@ expire_flapping(NowTime, #{duration := Duration, banned_interval := Interval}) -
|
|||
case ets:select(?FLAPPING_TAB,
|
||||
[{#flapping{started_at = '$1', banned_at = undefined, _ = '_'},
|
||||
[{'<', '$1', NowTime-Duration}], ['$_']},
|
||||
{#flapping{client_id = {banned, '_'}, banned_at = '$1', _ = '_'},
|
||||
{#flapping{clientid = {banned, '_'}, banned_at = '$1', _ = '_'},
|
||||
[{'<', '$1', NowTime-Interval}], ['$_']}]) of
|
||||
[] -> ok;
|
||||
Flappings ->
|
||||
lists:foreach(fun(Flapping = #flapping{client_id = {banned, ClientId}}) ->
|
||||
lists:foreach(fun(Flapping = #flapping{clientid = {banned, ClientId}}) ->
|
||||
ets:delete_object(?FLAPPING_TAB, Flapping),
|
||||
alarm_handler:clear_alarm({flapping_detected, ClientId});
|
||||
(_) -> ok
|
||||
|
|
|
@ -180,7 +180,7 @@ parse_packet(#mqtt_packet_header{type = ?CONNECT}, FrameBin, _Options) ->
|
|||
will_retain = bool(WillRetain),
|
||||
keepalive = KeepAlive,
|
||||
properties = Properties,
|
||||
client_id = ClientId},
|
||||
clientid = ClientId},
|
||||
{ConnPacket1, Rest5} = parse_will_message(ConnPacket, Rest4),
|
||||
{Username, Rest6} = parse_utf8_string(Rest5, bool(UsernameFlag)),
|
||||
{Passsword, <<>>} = parse_utf8_string(Rest6, bool(PasswordFlag)),
|
||||
|
@ -435,7 +435,7 @@ serialize_variable(#mqtt_packet_connect{
|
|||
will_retain = WillRetain,
|
||||
keepalive = KeepAlive,
|
||||
properties = Properties,
|
||||
client_id = ClientId,
|
||||
clientid = ClientId,
|
||||
will_props = WillProps,
|
||||
will_topic = WillTopic,
|
||||
will_payload = WillPayload,
|
||||
|
|
|
@ -38,7 +38,7 @@
|
|||
|
||||
%% Configs
|
||||
-export([ set_metadata_peername/1
|
||||
, set_metadata_client_id/1
|
||||
, set_metadata_clientid/1
|
||||
, set_proc_metadata/1
|
||||
, set_primary_log_level/1
|
||||
, set_log_handler_level/2
|
||||
|
@ -121,11 +121,11 @@ critical(Format, Args) ->
|
|||
critical(Metadata, Format, Args) when is_map(Metadata) ->
|
||||
logger:critical(Format, Args, Metadata).
|
||||
|
||||
-spec(set_metadata_client_id(emqx_types:client_id()) -> ok).
|
||||
set_metadata_client_id(<<>>) ->
|
||||
-spec(set_metadata_clientid(emqx_types:clientid()) -> ok).
|
||||
set_metadata_clientid(<<>>) ->
|
||||
ok;
|
||||
set_metadata_client_id(ClientId) ->
|
||||
set_proc_metadata(#{client_id => ClientId}).
|
||||
set_metadata_clientid(ClientId) ->
|
||||
set_proc_metadata(#{clientid => ClientId}).
|
||||
|
||||
-spec(set_metadata_peername(peername_str()) -> ok).
|
||||
set_metadata_peername(Peername) ->
|
||||
|
|
|
@ -71,13 +71,13 @@
|
|||
make(Topic, Payload) ->
|
||||
make(undefined, Topic, Payload).
|
||||
|
||||
-spec(make(atom() | emqx_types:client_id(),
|
||||
-spec(make(atom() | emqx_types:clientid(),
|
||||
emqx_topic:topic(),
|
||||
emqx_types:payload()) -> emqx_types:message()).
|
||||
make(From, Topic, Payload) ->
|
||||
make(From, ?QOS_0, Topic, Payload).
|
||||
|
||||
-spec(make(atom() | emqx_types:client_id(),
|
||||
-spec(make(atom() | emqx_types:clientid(),
|
||||
emqx_types:qos(),
|
||||
emqx_topic:topic(),
|
||||
emqx_types:payload()) -> emqx_types:message()).
|
||||
|
|
|
@ -61,7 +61,7 @@ all_rules() ->
|
|||
%%--------------------------------------------------------------------
|
||||
|
||||
%% @doc Check ACL
|
||||
-spec(check_acl(emqx_types:client_info(), emqx_types:pubsub(), emqx_topic:topic(),
|
||||
-spec(check_acl(emqx_types:clientinfo(), emqx_types:pubsub(), emqx_topic:topic(),
|
||||
emqx_access_rule:acl_result(), acl_rules())
|
||||
-> {ok, allow} | {ok, deny} | ok).
|
||||
check_acl(Client, PubSub, Topic, _AclResult, Rules) ->
|
||||
|
|
|
@ -88,8 +88,8 @@ on_client_disconnected(ClientInfo, Reason, ConnInfo, Env) ->
|
|||
?LOG(error, "Failed to encode 'disconnected' presence: ~p", [Presence])
|
||||
end.
|
||||
|
||||
clientid(#{client_id := undefined}, #{client_id := ClientId}) -> ClientId;
|
||||
clientid(#{client_id := ClientId}, _ConnInfo) -> ClientId.
|
||||
clientid(#{clientid := undefined}, #{clientid := ClientId}) -> ClientId;
|
||||
clientid(#{clientid := ClientId}, _ConnInfo) -> ClientId.
|
||||
|
||||
username(#{username := undefined}, #{username := Username}) -> Username;
|
||||
username(#{username := Username}, _ConnInfo) -> Username.
|
||||
|
|
|
@ -36,7 +36,7 @@
|
|||
load(Topics) ->
|
||||
emqx_hooks:add('client.connected', {?MODULE, on_client_connected, [Topics]}).
|
||||
|
||||
on_client_connected(#{client_id := ClientId,
|
||||
on_client_connected(#{clientid := ClientId,
|
||||
username := Username}, ?RC_SUCCESS, _ConnInfo, Topics) ->
|
||||
Replace = fun(Topic) ->
|
||||
rep(<<"%u">>, Username, rep(<<"%c">>, ClientId, Topic))
|
||||
|
|
|
@ -66,7 +66,7 @@ unmount(MountPoint, Msg = #message{topic = Topic}) ->
|
|||
-spec(replvar(maybe(mountpoint()), map()) -> maybe(mountpoint())).
|
||||
replvar(undefined, _Vars) ->
|
||||
undefined;
|
||||
replvar(MountPoint, #{client_id := ClientId, username := Username}) ->
|
||||
replvar(MountPoint, #{clientid := ClientId, username := Username}) ->
|
||||
lists:foldl(fun feed_var/2, MountPoint,
|
||||
[{<<"%c">>, ClientId}, {<<"%u">>, Username}]).
|
||||
|
||||
|
|
|
@ -181,16 +181,16 @@ check_proto_ver(#mqtt_packet_connect{proto_ver = Ver,
|
|||
|
||||
%% MQTT3.1 does not allow null clientId
|
||||
check_client_id(#mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V3,
|
||||
client_id = <<>>}, _Opts) ->
|
||||
clientid = <<>>}, _Opts) ->
|
||||
{error, ?RC_CLIENT_IDENTIFIER_NOT_VALID};
|
||||
%% Issue#599: Null clientId and clean_start = false
|
||||
check_client_id(#mqtt_packet_connect{client_id = <<>>,
|
||||
check_client_id(#mqtt_packet_connect{clientid = <<>>,
|
||||
clean_start = false}, _Opts) ->
|
||||
{error, ?RC_CLIENT_IDENTIFIER_NOT_VALID};
|
||||
check_client_id(#mqtt_packet_connect{client_id = <<>>,
|
||||
check_client_id(#mqtt_packet_connect{clientid = <<>>,
|
||||
clean_start = true}, _Opts) ->
|
||||
ok;
|
||||
check_client_id(#mqtt_packet_connect{client_id = ClientId},
|
||||
check_client_id(#mqtt_packet_connect{clientid = ClientId},
|
||||
_Opts = #{max_clientid_len := MaxLen}) ->
|
||||
case (1 =< (Len = byte_size(ClientId))) andalso (Len =< MaxLen) of
|
||||
true -> ok;
|
||||
|
@ -240,8 +240,8 @@ validate_topic_filters(TopicFilters) ->
|
|||
end, TopicFilters).
|
||||
|
||||
%% @doc Publish Packet to Message.
|
||||
-spec(to_message(emqx_types:client_info(), emqx_ypes:packet()) -> emqx_types:message()).
|
||||
to_message(#{client_id := ClientId, username := Username, peerhost := PeerHost},
|
||||
-spec(to_message(emqx_types:clientinfo(), emqx_ypes:packet()) -> emqx_types:message()).
|
||||
to_message(#{clientid := ClientId, username := Username, peerhost := PeerHost},
|
||||
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBLISH,
|
||||
retain = Retain,
|
||||
qos = QoS,
|
||||
|
@ -257,7 +257,7 @@ to_message(#{client_id := ClientId, username := Username, peerhost := PeerHost},
|
|||
-spec(will_msg(#mqtt_packet_connect{}) -> emqx_types:message()).
|
||||
will_msg(#mqtt_packet_connect{will_flag = false}) ->
|
||||
undefined;
|
||||
will_msg(#mqtt_packet_connect{client_id = ClientId,
|
||||
will_msg(#mqtt_packet_connect{clientid = ClientId,
|
||||
username = Username,
|
||||
will_retain = Retain,
|
||||
will_qos = QoS,
|
||||
|
@ -304,7 +304,7 @@ format_variable(#mqtt_packet_connect{
|
|||
will_flag = WillFlag,
|
||||
clean_start = CleanStart,
|
||||
keepalive = KeepAlive,
|
||||
client_id = ClientId,
|
||||
clientid = ClientId,
|
||||
will_topic = WillTopic,
|
||||
will_payload = WillPayload,
|
||||
username = Username,
|
||||
|
|
|
@ -28,16 +28,16 @@
|
|||
, stop_trace/1
|
||||
]).
|
||||
|
||||
-type(trace_who() :: {client_id | topic, binary() | list()}).
|
||||
-type(trace_who() :: {clientid | topic, binary() | list()}).
|
||||
|
||||
-define(TRACER, ?MODULE).
|
||||
-define(FORMAT, {emqx_logger_formatter,
|
||||
#{template =>
|
||||
[time," [",level,"] ",
|
||||
{client_id,
|
||||
{clientid,
|
||||
[{peername,
|
||||
[client_id,"@",peername," "],
|
||||
[client_id, " "]}],
|
||||
[clientid,"@",peername," "],
|
||||
[clientid, " "]}],
|
||||
[{peername,
|
||||
[peername," "],
|
||||
[]}]},
|
||||
|
@ -45,7 +45,7 @@
|
|||
-define(TOPIC_TRACE_ID(T), "trace_topic_"++T).
|
||||
-define(CLIENT_TRACE_ID(C), "trace_clientid_"++C).
|
||||
-define(TOPIC_TRACE(T), {topic,T}).
|
||||
-define(CLIENT_TRACE(C), {client_id,C}).
|
||||
-define(CLIENT_TRACE(C), {clientid,C}).
|
||||
|
||||
-define(is_log_level(L),
|
||||
L =:= emergency orelse
|
||||
|
@ -67,7 +67,7 @@ trace(publish, #message{from = From, topic = Topic, payload = Payload})
|
|||
when is_binary(From); is_atom(From) ->
|
||||
emqx_logger:info(#{topic => Topic, mfa => {?MODULE, ?FUNCTION_NAME, ?FUNCTION_ARITY} }, "PUBLISH to ~s: ~p", [Topic, Payload]).
|
||||
|
||||
%% @doc Start to trace client_id or topic.
|
||||
%% @doc Start to trace clientid or topic.
|
||||
-spec(start_trace(trace_who(), logger:level(), string()) -> ok | {error, term()}).
|
||||
start_trace(Who, all, LogFile) ->
|
||||
start_trace(Who, debug, LogFile);
|
||||
|
@ -87,7 +87,7 @@ start_trace(Who, Level, LogFile) ->
|
|||
false -> {error, {invalid_log_level, Level}}
|
||||
end.
|
||||
|
||||
%% @doc Stop tracing client_id or topic.
|
||||
%% @doc Stop tracing clientid or topic.
|
||||
-spec(stop_trace(trace_who()) -> ok | {error, term()}).
|
||||
stop_trace(Who) ->
|
||||
uninstall_trance_handler(Who).
|
||||
|
|
|
@ -32,8 +32,8 @@
|
|||
]).
|
||||
|
||||
-export_type([ conninfo/0
|
||||
, client_info/0
|
||||
, client_id/0
|
||||
, clientinfo/0
|
||||
, clientid/0
|
||||
, username/0
|
||||
, password/0
|
||||
, peerhost/0
|
||||
|
@ -78,6 +78,7 @@
|
|||
]).
|
||||
|
||||
-export_type([ caps/0
|
||||
, attrs/0
|
||||
, infos/0
|
||||
, stats/0
|
||||
]).
|
||||
|
@ -97,16 +98,27 @@
|
|||
|
||||
-type(socktype() :: tcp | udp | ssl | proxy | atom()).
|
||||
-type(conninfo() :: #{socktype := socktype(),
|
||||
peername := peername(),
|
||||
sockname := peername(),
|
||||
peername := peername(),
|
||||
peercert := esockd_peercert:peercert(),
|
||||
conn_mod := module(),
|
||||
proto_name := binary(),
|
||||
proto_ver := ver(),
|
||||
clean_start := boolean(),
|
||||
clientid := clientid(),
|
||||
username := username(),
|
||||
conn_props := properties(),
|
||||
connected := boolean(),
|
||||
connected_at := erlang:timestamp(),
|
||||
keepalive := 0..16#FFFF,
|
||||
receive_maximum := non_neg_integer(),
|
||||
expiry_interval := non_neg_integer(),
|
||||
atom() => term()
|
||||
}).
|
||||
-type(client_info() :: #{zone := zone(),
|
||||
-type(clientinfo() :: #{zone := zone(),
|
||||
protocol := protocol(),
|
||||
peerhost := peerhost(),
|
||||
client_id := client_id(),
|
||||
clientid := clientid(),
|
||||
username := username(),
|
||||
peercert := esockd_peercert:peercert(),
|
||||
is_bridge := boolean(),
|
||||
|
@ -118,7 +130,7 @@
|
|||
anonymous => boolean(),
|
||||
atom() => term()
|
||||
}).
|
||||
-type(client_id() :: binary()|atom()).
|
||||
-type(clientid() :: binary()|atom()).
|
||||
-type(username() :: maybe(binary())).
|
||||
-type(password() :: maybe(binary())).
|
||||
-type(peerhost() :: inet:ip_address()).
|
||||
|
@ -167,6 +179,7 @@
|
|||
-type(command() :: #command{}).
|
||||
|
||||
-type(caps() :: emqx_mqtt_caps:caps()).
|
||||
-type(attrs() :: #{atom() => term()}).
|
||||
-type(infos() :: #{atom() => term()}).
|
||||
-type(stats() :: #{atom() => non_neg_integer()|stats()}).
|
||||
|
||||
|
|
|
@ -113,7 +113,7 @@ t_reload_acl(_) ->
|
|||
|
||||
t_check_acl_1(_) ->
|
||||
Client = #{zone => external,
|
||||
client_id => <<"client1">>,
|
||||
clientid => <<"client1">>,
|
||||
username => <<"testuser">>
|
||||
},
|
||||
allow = ?AC:check_acl(Client, subscribe, <<"users/testuser/1">>),
|
||||
|
@ -124,14 +124,14 @@ t_check_acl_1(_) ->
|
|||
|
||||
t_check_acl_2(_) ->
|
||||
Client = #{zone => external,
|
||||
client_id => <<"client2">>,
|
||||
clientid => <<"client2">>,
|
||||
username => <<"xyz">>
|
||||
},
|
||||
deny = ?AC:check_acl(Client, subscribe, <<"a/b/c">>).
|
||||
|
||||
t_acl_cache_basic(_) ->
|
||||
Client = #{zone => external,
|
||||
client_id => <<"client1">>,
|
||||
clientid => <<"client1">>,
|
||||
username => <<"testuser">>
|
||||
},
|
||||
not_found = ?CACHE:get_acl_cache(subscribe, <<"users/testuser/1">>),
|
||||
|
@ -146,7 +146,7 @@ t_acl_cache_basic(_) ->
|
|||
t_acl_cache_expiry(_) ->
|
||||
application:set_env(emqx, acl_cache_ttl, 100),
|
||||
Client = #{zone => external,
|
||||
client_id => <<"client1">>,
|
||||
clientid => <<"client1">>,
|
||||
username => <<"testuser">>
|
||||
},
|
||||
allow = ?AC:check_acl(Client, subscribe, <<"clients/client1">>),
|
||||
|
@ -157,7 +157,7 @@ t_acl_cache_expiry(_) ->
|
|||
t_acl_cache_full(_) ->
|
||||
application:set_env(emqx, acl_cache_max_size, 1),
|
||||
Client = #{zone => external,
|
||||
client_id => <<"client1">>,
|
||||
clientid => <<"client1">>,
|
||||
username => <<"testuser">>
|
||||
},
|
||||
allow = ?AC:check_acl(Client, subscribe, <<"users/testuser/1">>),
|
||||
|
@ -173,7 +173,7 @@ t_acl_cache_cleanup(_) ->
|
|||
application:set_env(emqx, acl_cache_ttl, 100),
|
||||
application:set_env(emqx, acl_cache_max_size, 2),
|
||||
Client = #{zone => external,
|
||||
client_id => <<"client1">>,
|
||||
clientid => <<"client1">>,
|
||||
username => <<"testuser">>
|
||||
},
|
||||
allow = ?AC:check_acl(Client, subscribe, <<"users/testuser/1">>),
|
||||
|
@ -345,12 +345,12 @@ t_compile_rule(_) ->
|
|||
|
||||
t_match_rule(_) ->
|
||||
ClientInfo1 = #{zone => external,
|
||||
client_id => <<"testClient">>,
|
||||
clientid => <<"testClient">>,
|
||||
username => <<"TestUser">>,
|
||||
peerhost => {127,0,0,1}
|
||||
},
|
||||
ClientInfo2 = #{zone => external,
|
||||
client_id => <<"testClient">>,
|
||||
clientid => <<"testClient">>,
|
||||
username => <<"TestUser">>,
|
||||
peerhost => {192,168,0,10}
|
||||
},
|
||||
|
|
|
@ -37,7 +37,7 @@ end_per_suite(_Config) ->
|
|||
ekka_mnesia:delete_schema().
|
||||
|
||||
t_add_delete(_) ->
|
||||
Banned = #banned{who = {client_id, <<"TestClient">>},
|
||||
Banned = #banned{who = {clientid, <<"TestClient">>},
|
||||
reason = <<"test">>,
|
||||
by = <<"banned suite">>,
|
||||
desc = <<"test">>,
|
||||
|
@ -45,27 +45,27 @@ t_add_delete(_) ->
|
|||
},
|
||||
ok = emqx_banned:add(Banned),
|
||||
?assertEqual(1, emqx_banned:info(size)),
|
||||
ok = emqx_banned:delete({client_id, <<"TestClient">>}),
|
||||
ok = emqx_banned:delete({clientid, <<"TestClient">>}),
|
||||
?assertEqual(0, emqx_banned:info(size)).
|
||||
|
||||
t_check(_) ->
|
||||
ok = emqx_banned:add(#banned{who = {client_id, <<"BannedClient">>}}),
|
||||
ok = emqx_banned:add(#banned{who = {clientid, <<"BannedClient">>}}),
|
||||
ok = emqx_banned:add(#banned{who = {username, <<"BannedUser">>}}),
|
||||
ok = emqx_banned:add(#banned{who = {ipaddr, {192,168,0,1}}}),
|
||||
?assertEqual(3, emqx_banned:info(size)),
|
||||
ClientInfo1 = #{client_id => <<"BannedClient">>,
|
||||
ClientInfo1 = #{clientid => <<"BannedClient">>,
|
||||
username => <<"user">>,
|
||||
peerhost => {127,0,0,1}
|
||||
},
|
||||
ClientInfo2 = #{client_id => <<"client">>,
|
||||
ClientInfo2 = #{clientid => <<"client">>,
|
||||
username => <<"BannedUser">>,
|
||||
peerhost => {127,0,0,1}
|
||||
},
|
||||
ClientInfo3 = #{client_id => <<"client">>,
|
||||
ClientInfo3 = #{clientid => <<"client">>,
|
||||
username => <<"user">>,
|
||||
peerhost => {192,168,0,1}
|
||||
},
|
||||
ClientInfo4 = #{client_id => <<"client">>,
|
||||
ClientInfo4 = #{clientid => <<"client">>,
|
||||
username => <<"user">>,
|
||||
peerhost => {127,0,0,1}
|
||||
},
|
||||
|
@ -73,7 +73,7 @@ t_check(_) ->
|
|||
?assert(emqx_banned:check(ClientInfo2)),
|
||||
?assert(emqx_banned:check(ClientInfo3)),
|
||||
?assertNot(emqx_banned:check(ClientInfo4)),
|
||||
ok = emqx_banned:delete({client_id, <<"BannedClient">>}),
|
||||
ok = emqx_banned:delete({clientid, <<"BannedClient">>}),
|
||||
ok = emqx_banned:delete({username, <<"BannedUser">>}),
|
||||
ok = emqx_banned:delete({ipaddr, {192,168,0,1}}),
|
||||
?assertNot(emqx_banned:check(ClientInfo1)),
|
||||
|
@ -84,7 +84,7 @@ t_check(_) ->
|
|||
|
||||
t_unused(_) ->
|
||||
{ok, Banned} = emqx_banned:start_link(),
|
||||
ok = emqx_banned:add(#banned{who = {client_id, <<"BannedClient">>},
|
||||
ok = emqx_banned:add(#banned{who = {clientid, <<"BannedClient">>},
|
||||
until = erlang:system_time(second)
|
||||
}),
|
||||
?assertEqual(ignored, gen_server:call(Banned, unexpected_req)),
|
||||
|
|
|
@ -63,17 +63,17 @@ t_get_set_chan_stats(_) ->
|
|||
|
||||
t_open_session(_) ->
|
||||
ClientInfo = #{zone => external,
|
||||
client_id => <<"clientid">>,
|
||||
clientid => <<"clientid">>,
|
||||
username => <<"username">>,
|
||||
peerhost => {127,0,0,1}},
|
||||
ConnInfo = #{peername => {{127,0,0,1}, 5000},
|
||||
receive_maximum => 100},
|
||||
{ok, #{session := Session1, present := false}}
|
||||
= emqx_cm:open_session(true, ClientInfo, ConnInfo),
|
||||
?assertEqual(100, emqx_session:info(max_inflight, Session1)),
|
||||
?assertEqual(100, emqx_session:info(inflight_max, Session1)),
|
||||
{ok, #{session := Session2, present := false}}
|
||||
= emqx_cm:open_session(false, ClientInfo, ConnInfo),
|
||||
?assertEqual(100, emqx_session:info(max_inflight, Session2)).
|
||||
?assertEqual(100, emqx_session:info(inflight_max, Session2)).
|
||||
|
||||
t_discard_session(_) ->
|
||||
ok = emqx_cm:discard_session(<<"clientid">>).
|
||||
|
|
|
@ -41,7 +41,7 @@ end_per_suite(_Config) ->
|
|||
|
||||
t_detect_check(_) ->
|
||||
ClientInfo = #{zone => external,
|
||||
client_id => <<"clientid">>,
|
||||
clientid => <<"clientid">>,
|
||||
peerhost => {127,0,0,1}
|
||||
},
|
||||
false = emqx_flapping:detect(ClientInfo),
|
||||
|
|
|
@ -147,7 +147,7 @@ prop_serialize_parse_connect() ->
|
|||
Packet = ?CONNECT_PACKET(#mqtt_packet_connect{
|
||||
proto_name = ProtoName,
|
||||
proto_ver = ProtoVer,
|
||||
client_id = <<"clientId">>,
|
||||
clientid = <<"clientId">>,
|
||||
will_qos = ?QOS_1,
|
||||
will_flag = true,
|
||||
will_retain = true,
|
||||
|
@ -167,7 +167,7 @@ t_serialize_parse_v3_connect(_) ->
|
|||
Packet = ?CONNECT_PACKET(
|
||||
#mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V3,
|
||||
proto_name = <<"MQIsdp">>,
|
||||
client_id = <<"mosqpub/10451-iMac.loca">>,
|
||||
clientid = <<"mosqpub/10451-iMac.loca">>,
|
||||
clean_start = true,
|
||||
keepalive = 60
|
||||
}),
|
||||
|
@ -180,7 +180,7 @@ t_serialize_parse_v4_connect(_) ->
|
|||
Packet = ?CONNECT_PACKET(
|
||||
#mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V4,
|
||||
proto_name = <<"MQTT">>,
|
||||
client_id = <<"mosqpub/10451-iMac.loca">>,
|
||||
clientid = <<"mosqpub/10451-iMac.loca">>,
|
||||
clean_start = true,
|
||||
keepalive = 60
|
||||
}),
|
||||
|
@ -213,7 +213,7 @@ t_serialize_parse_v5_connect(_) ->
|
|||
proto_ver = ?MQTT_PROTO_V5,
|
||||
is_bridge = false,
|
||||
clean_start = true,
|
||||
client_id = <<>>,
|
||||
clientid = <<>>,
|
||||
will_flag = true,
|
||||
will_qos = ?QOS_1,
|
||||
will_retain = false,
|
||||
|
@ -231,7 +231,7 @@ t_serialize_parse_connect_without_clientid(_) ->
|
|||
Bin = <<16,12,0,4,77,81,84,84,4,2,0,60,0,0>>,
|
||||
Packet = ?CONNECT_PACKET(#mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V4,
|
||||
proto_name = <<"MQTT">>,
|
||||
client_id = <<>>,
|
||||
clientid = <<>>,
|
||||
clean_start = true,
|
||||
keepalive = 60
|
||||
}),
|
||||
|
@ -246,7 +246,7 @@ t_serialize_parse_connect_with_will(_) ->
|
|||
Packet = #mqtt_packet{header = #mqtt_packet_header{type = ?CONNECT},
|
||||
variable = #mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V3,
|
||||
proto_name = <<"MQIsdp">>,
|
||||
client_id = <<"mosqpub/10452-iMac.loca">>,
|
||||
clientid = <<"mosqpub/10452-iMac.loca">>,
|
||||
clean_start = true,
|
||||
keepalive = 60,
|
||||
will_retain = false,
|
||||
|
@ -267,7 +267,7 @@ t_serialize_parse_bridge_connect(_) ->
|
|||
67,58,50,57,58,50,66,58,55,55,58,53,50,47,115,116,97,116,101,0,1,48>>,
|
||||
Topic = <<"$SYS/broker/connection/C_00:0C:29:2B:77:52/state">>,
|
||||
Packet = #mqtt_packet{header = #mqtt_packet_header{type = ?CONNECT},
|
||||
variable = #mqtt_packet_connect{client_id = <<"C_00:0C:29:2B:77:52">>,
|
||||
variable = #mqtt_packet_connect{clientid = <<"C_00:0C:29:2B:77:52">>,
|
||||
proto_ver = 16#03,
|
||||
proto_name = <<"MQIsdp">>,
|
||||
is_bridge = true,
|
||||
|
|
|
@ -56,12 +56,12 @@ t_replvar(_) ->
|
|||
?assertEqual(undefined, replvar(undefined, #{})),
|
||||
?assertEqual(<<"mount/user/clientid/">>,
|
||||
replvar(<<"mount/%u/%c/">>,
|
||||
#{client_id => <<"clientid">>,
|
||||
#{clientid => <<"clientid">>,
|
||||
username => <<"user">>
|
||||
})),
|
||||
?assertEqual(<<"mount/%u/clientid/">>,
|
||||
replvar(<<"mount/%u/%c/">>,
|
||||
#{client_id => <<"clientid">>,
|
||||
#{clientid => <<"clientid">>,
|
||||
username => undefined
|
||||
})).
|
||||
|
||||
|
|
|
@ -42,8 +42,8 @@ t_message_expiry_interval_2(_) ->
|
|||
emqtt:stop(ClientA).
|
||||
|
||||
message_expiry_interval_init() ->
|
||||
{ok, ClientA} = emqtt:start_link([{proto_ver,v5}, {client_id, <<"client-a">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, ClientB} = emqtt:start_link([{proto_ver,v5}, {client_id, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, ClientA} = emqtt:start_link([{proto_ver,v5}, {clientid, <<"client-a">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, ClientB} = emqtt:start_link([{proto_ver,v5}, {clientid, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, _} = emqtt:connect(ClientA),
|
||||
{ok, _} = emqtt:connect(ClientB),
|
||||
%% subscribe and disconnect client-b
|
||||
|
@ -58,7 +58,7 @@ message_expiry_interval_exipred(ClientA, QoS) ->
|
|||
ct:sleep(1500),
|
||||
|
||||
%% resume the session for client-b
|
||||
{ok, ClientB1} = emqtt:start_link([{proto_ver,v5}, {client_id, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, ClientB1} = emqtt:start_link([{proto_ver,v5}, {clientid, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, _} = emqtt:connect(ClientB1),
|
||||
|
||||
%% verify client-b could not receive the publish message
|
||||
|
@ -78,7 +78,7 @@ message_expiry_interval_not_exipred(ClientA, QoS) ->
|
|||
%% wait for 1s and then resume the session for client-b, the message should not expires
|
||||
%% as Message-Expiry-Interval = 20s
|
||||
ct:sleep(1000),
|
||||
{ok, ClientB1} = emqtt:start_link([{proto_ver,v5}, {client_id, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, ClientB1} = emqtt:start_link([{proto_ver,v5}, {clientid, <<"client-b">>}, {clean_start, false},{properties, #{'Session-Expiry-Interval' => 360}}]),
|
||||
{ok, _} = emqtt:connect(ClientB1),
|
||||
|
||||
%% verify client-b could receive the publish message and the Message-Expiry-Interval is set
|
||||
|
|
|
@ -112,11 +112,11 @@ t_check_connect(_) ->
|
|||
|
||||
ConnPkt2 = #mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V3,
|
||||
proto_name = <<"MQIsdp">>,
|
||||
client_id = <<>>
|
||||
clientid = <<>>
|
||||
},
|
||||
{error, ?RC_CLIENT_IDENTIFIER_NOT_VALID} = emqx_packet:check(ConnPkt2, Opts),
|
||||
|
||||
ConnPkt3 = #mqtt_packet_connect{client_id = <<"123456">>},
|
||||
ConnPkt3 = #mqtt_packet_connect{clientid = <<"123456">>},
|
||||
{error, ?RC_CLIENT_IDENTIFIER_NOT_VALID} = emqx_packet:check(ConnPkt3, Opts),
|
||||
|
||||
ConnPkt4 = #mqtt_packet_connect{will_flag = true,
|
||||
|
@ -152,7 +152,7 @@ t_from_to_message(_) ->
|
|||
packet_id = 10,
|
||||
properties = #{}},
|
||||
payload = <<"payload">>},
|
||||
MsgFromPkt = emqx_packet:to_message(#{client_id => <<"clientid">>,
|
||||
MsgFromPkt = emqx_packet:to_message(#{clientid => <<"clientid">>,
|
||||
username => <<"test">>,
|
||||
peerhost => {127,0,0,1}}, Pkt),
|
||||
?assertEqual(ExpectedMsg2, MsgFromPkt#message{id = emqx_message:id(ExpectedMsg),
|
||||
|
@ -161,7 +161,7 @@ t_from_to_message(_) ->
|
|||
|
||||
t_will_msg(_) ->
|
||||
Pkt = #mqtt_packet_connect{will_flag = true,
|
||||
client_id = <<"clientid">>,
|
||||
clientid = <<"clientid">>,
|
||||
username = "test",
|
||||
will_retain = true,
|
||||
will_qos = ?QOS_2,
|
||||
|
|
|
@ -35,18 +35,19 @@ end_per_suite(_Config) ->
|
|||
|
||||
t_start_traces(_Config) ->
|
||||
{ok, T} = emqtt:start_link([{host, "localhost"},
|
||||
{client_id, <<"client">>},
|
||||
{clientid, <<"client">>},
|
||||
{username, <<"testuser">>},
|
||||
{password, <<"pass">>}]),
|
||||
{password, <<"pass">>}
|
||||
]),
|
||||
emqtt:connect(T),
|
||||
|
||||
%% Start tracing
|
||||
emqx_logger:set_log_level(error),
|
||||
{error, _} = emqx_tracer:start_trace({client_id, <<"client">>}, debug, "tmp/client.log"),
|
||||
{error, _} = emqx_tracer:start_trace({clientid, <<"client">>}, debug, "tmp/client.log"),
|
||||
emqx_logger:set_log_level(debug),
|
||||
ok = emqx_tracer:start_trace({client_id, <<"client">>}, debug, "tmp/client.log"),
|
||||
ok = emqx_tracer:start_trace({client_id, <<"client2">>}, all, "tmp/client2.log"),
|
||||
{error, {invalid_log_level, bad_level}} = emqx_tracer:start_trace({client_id, <<"client3">>}, bad_level, "tmp/client3.log"),
|
||||
ok = emqx_tracer:start_trace({clientid, <<"client">>}, debug, "tmp/client.log"),
|
||||
ok = emqx_tracer:start_trace({clientid, <<"client2">>}, all, "tmp/client2.log"),
|
||||
{error, {invalid_log_level, bad_level}} = emqx_tracer:start_trace({clientid, <<"client3">>}, bad_level, "tmp/client3.log"),
|
||||
ok = emqx_tracer:start_trace({topic, <<"a/#">>}, all, "tmp/topic_trace.log"),
|
||||
ct:sleep(100),
|
||||
|
||||
|
@ -56,8 +57,8 @@ t_start_traces(_Config) ->
|
|||
?assert(filelib:is_regular("tmp/topic_trace.log")),
|
||||
|
||||
%% Get current traces
|
||||
?assertEqual([{{client_id,"client"},{debug,"tmp/client.log"}},
|
||||
{{client_id,"client2"},{debug,"tmp/client2.log"}},
|
||||
?assertEqual([{{clientid,"client"},{debug,"tmp/client.log"}},
|
||||
{{clientid,"client2"},{debug,"tmp/client2.log"}},
|
||||
{{topic,"a/#"},{debug,"tmp/topic_trace.log"}}], emqx_tracer:lookup_traces()),
|
||||
|
||||
%% set the overall log level to debug
|
||||
|
@ -73,8 +74,8 @@ t_start_traces(_Config) ->
|
|||
?assert(filelib:file_size("tmp/client2.log") == 0),
|
||||
|
||||
%% Stop tracing
|
||||
ok = emqx_tracer:stop_trace({client_id, <<"client">>}),
|
||||
ok = emqx_tracer:stop_trace({client_id, <<"client2">>}),
|
||||
ok = emqx_tracer:stop_trace({clientid, <<"client">>}),
|
||||
ok = emqx_tracer:stop_trace({clientid, <<"client2">>}),
|
||||
ok = emqx_tracer:stop_trace({topic, <<"a/#">>}),
|
||||
emqtt:disconnect(T),
|
||||
|
||||
|
|
Loading…
Reference in New Issue