diff --git a/apps/emqx_management/src/emqx_mgmt.erl b/apps/emqx_management/src/emqx_mgmt.erl index bfba34603..93f24cd37 100644 --- a/apps/emqx_management/src/emqx_mgmt.erl +++ b/apps/emqx_management/src/emqx_mgmt.erl @@ -22,6 +22,9 @@ -include_lib("emqx/include/emqx.hrl"). -include_lib("emqx/include/emqx_mqtt.hrl"). +-elvis([{elvis_style, invalid_dynamic_call, #{ignore => [emqx_mgmt]}}]). +-elvis([{elvis_style, god_modules, #{ignore => [emqx_mgmt]}}]). + %% Nodes and Brokers API -export([ list_nodes/0 , lookup_node/1 @@ -49,6 +52,7 @@ , clean_acl_cache_all/1 , set_ratelimit_policy/2 , set_quota_policy/2 + , set_keepalive/2 ]). %% Internal funcs @@ -142,7 +146,8 @@ node_info(Node) when Node =:= node() -> memory_used => proplists:get_value(total, Memory), process_available => erlang:system_info(process_limit), process_used => erlang:system_info(process_count), - max_fds => proplists:get_value(max_fds, lists:usort(lists:flatten(erlang:system_info(check_io)))), + max_fds => proplists:get_value(max_fds, + lists:usort(lists:flatten(erlang:system_info(check_io)))), connections => ets:info(emqx_channel, size), node_status => 'Running', uptime => iolist_to_binary(proplists:get_value(uptime, BrokerInfo)), @@ -196,10 +201,12 @@ get_stats(Node) -> %%-------------------------------------------------------------------- lookup_client({clientid, ClientId}, FormatFun) -> - lists:append([lookup_client(Node, {clientid, ClientId}, FormatFun) || Node <- ekka_mnesia:running_nodes()]); + lists:append([lookup_client(Node, {clientid, ClientId}, FormatFun) + || Node <- ekka_mnesia:running_nodes()]); lookup_client({username, Username}, FormatFun) -> - lists:append([lookup_client(Node, {username, Username}, FormatFun) || Node <- ekka_mnesia:running_nodes()]). + lists:append([lookup_client(Node, {username, Username}, FormatFun) + || Node <- ekka_mnesia:running_nodes()]). lookup_client(Node, {clientid, ClientId}, {M,F}) when Node =:= node() -> lists:append(lists:map( @@ -222,10 +229,7 @@ lookup_client(Node, {username, Username}, FormatFun) -> kickout_client(ClientId) -> Results = [kickout_client(Node, ClientId) || Node <- ekka_mnesia:running_nodes()], - case lists:any(fun(Item) -> Item =:= ok end, Results) of - true -> ok; - false -> lists:last(Results) - end. + has_any_ok(Results). kickout_client(Node, ClientId) when Node =:= node() -> emqx_cm:kick_session(ClientId); @@ -238,10 +242,7 @@ list_acl_cache(ClientId) -> clean_acl_cache(ClientId) -> Results = [clean_acl_cache(Node, ClientId) || Node <- ekka_mnesia:running_nodes()], - case lists:any(fun(Item) -> Item =:= ok end, Results) of - true -> ok; - false -> lists:last(Results) - end. + has_any_ok(Results). clean_acl_cache(Node, ClientId) when Node =:= node() -> case emqx_cm:lookup_channels(ClientId) of @@ -273,6 +274,9 @@ set_ratelimit_policy(ClientId, Policy) -> set_quota_policy(ClientId, Policy) -> call_client(ClientId, {quota, Policy}). +set_keepalive(ClientId, Interval) -> + call_client(ClientId, {keepalive, Interval}). + %% @private call_client(ClientId, Req) -> Results = [call_client(Node, ClientId, Req) || Node <- ekka_mnesia:running_nodes()], @@ -281,7 +285,7 @@ call_client(ClientId, Req) -> end, Results), case Expected of [] -> {error, not_found}; - [Result|_] -> Result + [Result | _] -> Result end. %% @private @@ -313,7 +317,8 @@ list_subscriptions(Node) -> rpc_call(Node, list_subscriptions, [Node]). list_subscriptions_via_topic(Topic, FormatFun) -> - lists:append([list_subscriptions_via_topic(Node, Topic, FormatFun) || Node <- ekka_mnesia:running_nodes()]). + lists:append([list_subscriptions_via_topic(Node, Topic, FormatFun) + || Node <- ekka_mnesia:running_nodes()]). list_subscriptions_via_topic(Node, Topic, {M,F}) when Node =:= node() -> MatchSpec = [{{{'_', '$1'}, '_'}, [{'=:=','$1', Topic}], ['$_']}], @@ -436,7 +441,8 @@ list_listeners(Node) when Node =:= node() -> Http = lists:map(fun({Protocol, Opts}) -> #{protocol => Protocol, listen_on => proplists:get_value(port, Opts), - acceptors => maps:get(num_acceptors, proplists:get_value(transport_options, Opts, #{}), 0), + acceptors => maps:get(num_acceptors, + proplists:get_value(transport_options, Opts, #{}), 0), max_conns => proplists:get_value(max_connections, Opts), current_conns => proplists:get_value(all_connections, Opts), shutdown_count => []} @@ -483,9 +489,10 @@ add_duration_field(Alarms) -> add_duration_field([], _Now, Acc) -> Acc; -add_duration_field([Alarm = #{activated := true, activate_at := ActivateAt}| Rest], Now, Acc) -> +add_duration_field([Alarm = #{activated := true, activate_at := ActivateAt} | Rest], Now, Acc) -> add_duration_field(Rest, Now, [Alarm#{duration => Now - ActivateAt} | Acc]); -add_duration_field([Alarm = #{activated := false, activate_at := ActivateAt, deactivate_at := DeactivateAt}| Rest], Now, Acc) -> +add_duration_field([Alarm = #{activated := false, + activate_at := ActivateAt, deactivate_at := DeactivateAt} | Rest], Now, Acc) -> add_duration_field(Rest, Now, [Alarm#{duration => DeactivateAt - ActivateAt} | Acc]). %%-------------------------------------------------------------------- @@ -560,7 +567,7 @@ check_row_limit(Tables) -> check_row_limit([], _Limit) -> ok; -check_row_limit([Tab|Tables], Limit) -> +check_row_limit([Tab | Tables], Limit) -> case table_size(Tab) > Limit of true -> false; false -> check_row_limit(Tables, Limit) @@ -571,4 +578,8 @@ max_row_limit() -> table_size(Tab) -> ets:info(Tab, size). - +has_any_ok(Results) -> + case lists:any(fun(Item) -> Item =:= ok end, Results) of + true -> ok; + false -> lists:last(Results) + end. diff --git a/apps/emqx_management/src/emqx_mgmt_api_clients.erl b/apps/emqx_management/src/emqx_mgmt_api_clients.erl index 1ddd87a3d..5aacc7895 100644 --- a/apps/emqx_management/src/emqx_mgmt_api_clients.erl +++ b/apps/emqx_management/src/emqx_mgmt_api_clients.erl @@ -117,6 +117,12 @@ func => clean_quota, descr => "Clear the quota policy"}). +-rest_api(#{name => set_keepalive, + method => 'PUT', + path => "/clients/:bin:clientid/keepalive", + func => set_keepalive, + descr => "Set the client keepalive"}). + -import(emqx_mgmt_util, [ ntoa/1 , strftime/1 ]). @@ -130,23 +136,24 @@ , set_quota_policy/2 , clean_ratelimit/2 , clean_quota/2 + , set_keepalive/2 ]). -export([ query/3 , format_channel_info/1 ]). --define(query_fun, {?MODULE, query}). --define(format_fun, {?MODULE, format_channel_info}). +-define(QUERY_FUN, {?MODULE, query}). +-define(FORMAT_FUN, {?MODULE, format_channel_info}). list(Bindings, Params) when map_size(Bindings) == 0 -> fence(fun() -> - emqx_mgmt_api:cluster_query(Params, ?CLIENT_QS_SCHEMA, ?query_fun) + emqx_mgmt_api:cluster_query(Params, ?CLIENT_QS_SCHEMA, ?QUERY_FUN) end); list(#{node := Node}, Params) when Node =:= node() -> fence(fun() -> - emqx_mgmt_api:node_query(Node, Params, ?CLIENT_QS_SCHEMA, ?query_fun) + emqx_mgmt_api:node_query(Node, Params, ?CLIENT_QS_SCHEMA, ?QUERY_FUN) end); list(Bindings = #{node := Node}, Params) -> @@ -169,16 +176,20 @@ fence(Func) -> end. lookup(#{node := Node, clientid := ClientId}, _Params) -> - minirest:return({ok, emqx_mgmt:lookup_client(Node, {clientid, emqx_mgmt_util:urldecode(ClientId)}, ?format_fun)}); + minirest:return({ok, emqx_mgmt:lookup_client(Node, + {clientid, emqx_mgmt_util:urldecode(ClientId)}, ?FORMAT_FUN)}); lookup(#{clientid := ClientId}, _Params) -> - minirest:return({ok, emqx_mgmt:lookup_client({clientid, emqx_mgmt_util:urldecode(ClientId)}, ?format_fun)}); + minirest:return({ok, emqx_mgmt:lookup_client( + {clientid, emqx_mgmt_util:urldecode(ClientId)}, ?FORMAT_FUN)}); lookup(#{node := Node, username := Username}, _Params) -> - minirest:return({ok, emqx_mgmt:lookup_client(Node, {username, emqx_mgmt_util:urldecode(Username)}, ?format_fun)}); + minirest:return({ok, emqx_mgmt:lookup_client(Node, + {username, emqx_mgmt_util:urldecode(Username)}, ?FORMAT_FUN)}); lookup(#{username := Username}, _Params) -> - minirest:return({ok, emqx_mgmt:lookup_client({username, emqx_mgmt_util:urldecode(Username)}, ?format_fun)}). + minirest:return({ok, emqx_mgmt:lookup_client({username, + emqx_mgmt_util:urldecode(Username)}, ?FORMAT_FUN)}). kickout(#{clientid := ClientId}, _Params) -> case emqx_mgmt:kickout_client(emqx_mgmt_util:urldecode(ClientId)) of @@ -204,7 +215,7 @@ list_acl_cache(#{clientid := ClientId}, _Params) -> set_ratelimit_policy(#{clientid := ClientId}, Params) -> P = [{conn_bytes_in, proplists:get_value(<<"conn_bytes_in">>, Params)}, {conn_messages_in, proplists:get_value(<<"conn_messages_in">>, Params)}], - case [{K, parse_ratelimit_str(V)} || {K, V} <- P, V =/= undefined] of + case filter_ratelimit_params(P) of [] -> minirest:return(); Policy -> case emqx_mgmt:set_ratelimit_policy(emqx_mgmt_util:urldecode(ClientId), Policy) of @@ -223,7 +234,7 @@ clean_ratelimit(#{clientid := ClientId}, _Params) -> set_quota_policy(#{clientid := ClientId}, Params) -> P = [{conn_messages_routing, proplists:get_value(<<"conn_messages_routing">>, Params)}], - case [{K, parse_ratelimit_str(V)} || {K, V} <- P, V =/= undefined] of + case filter_ratelimit_params(P) of [] -> minirest:return(); Policy -> case emqx_mgmt:set_quota_policy(emqx_mgmt_util:urldecode(ClientId), Policy) of @@ -233,6 +244,7 @@ set_quota_policy(#{clientid := ClientId}, Params) -> end end. + clean_quota(#{clientid := ClientId}, _Params) -> case emqx_mgmt:set_quota_policy(emqx_mgmt_util:urldecode(ClientId), []) of ok -> minirest:return(); @@ -240,6 +252,19 @@ clean_quota(#{clientid := ClientId}, _Params) -> {error, Reason} -> minirest:return({error, ?ERROR1, Reason}) end. +set_keepalive(#{clientid := ClientId}, Params) -> + case proplists:get_value(<<"interval">>, Params) of + undefined -> + minirest:return({error, ?ERROR7, params_not_found}); + Interval0 -> + Interval = binary_to_integer(Interval0), + case emqx_mgmt:set_keepalive(emqx_mgmt_util:urldecode(ClientId), Interval) of + ok -> minirest:return(); + {error, not_found} -> minirest:return({error, ?ERROR12, not_found}); + {error, Reason} -> minirest:return({error, ?ERROR1, Reason}) + end + end. + %% @private %% S = 100,1s %% | 100KB, 1m @@ -266,7 +291,7 @@ format_channel_info({_Key, Info, Stats0}) -> ConnInfo = maps:get(conninfo, Info, #{}), Session = case maps:get(session, Info, #{}) of undefined -> #{}; - _Sess -> _Sess + Sess -> Sess end, SessCreated = maps:get(created_at, Session, maps:get(connected_at, ConnInfo)), Connected = case maps:get(conn_state, Info, connected) of @@ -306,7 +331,8 @@ format(Data) when is_map(Data)-> created_at => iolist_to_binary(strftime(CreatedAt div 1000))}, case maps:get(disconnected_at, Data, undefined) of undefined -> #{}; - DisconnectedAt -> #{disconnected_at => iolist_to_binary(strftime(DisconnectedAt div 1000))} + DisconnectedAt -> #{disconnected_at => + iolist_to_binary(strftime(DisconnectedAt div 1000))} end). format_acl_cache({{PubSub, Topic}, {AclResult, Timestamp}}) -> @@ -326,7 +352,8 @@ query({Qs, []}, Start, Limit) -> query({Qs, Fuzzy}, Start, Limit) -> Ms = qs2ms(Qs), MatchFun = match_fun(Ms, Fuzzy), - emqx_mgmt_api:traverse_table(emqx_channel_info, MatchFun, Start, Limit, fun format_channel_info/1). + emqx_mgmt_api:traverse_table(emqx_channel_info, MatchFun, + Start, Limit, fun format_channel_info/1). %%-------------------------------------------------------------------- %% Match funcs @@ -352,7 +379,7 @@ escape(B) when is_binary(B) -> run_fuzzy_match(_, []) -> true; -run_fuzzy_match(E = {_, #{clientinfo := ClientInfo}, _}, [{Key, _, RE}|Fuzzy]) -> +run_fuzzy_match(E = {_, #{clientinfo := ClientInfo}, _}, [{Key, _, RE} | Fuzzy]) -> Val = case maps:get(Key, ClientInfo, "") of undefined -> ""; V -> V @@ -406,6 +433,9 @@ ms(connected_at, X) -> ms(created_at, X) -> #{session => #{created_at => X}}. +filter_ratelimit_params(P) -> + [{K, parse_ratelimit_str(V)} || {K, V} <- P, V =/= undefined]. + %%-------------------------------------------------------------------- %% EUnits %%-------------------------------------------------------------------- diff --git a/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl b/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl index e45acfd42..f3bc42f3a 100644 --- a/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl +++ b/apps/emqx_management/test/emqx_mgmt_api_SUITE.erl @@ -29,6 +29,8 @@ -define(HOST, "http://127.0.0.1:8081/"). +-elvis([{elvis_style, line_length, disable}]). + -define(API_VERSION, "v4"). -define(BASE_PATH, "api"). @@ -76,30 +78,40 @@ t_alarms(_) -> ?assert(is_existing(alarm2, emqx_alarm:get_alarms(activated))), {ok, Return1} = request_api(get, api_path(["alarms/activated"]), auth_header_()), - ?assert(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return1))))), - ?assert(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return1))))), + ?assert(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return1))))), + ?assert(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return1))))), emqx_alarm:deactivate(alarm1), {ok, Return2} = request_api(get, api_path(["alarms"]), auth_header_()), - ?assert(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return2))))), - ?assert(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return2))))), + ?assert(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return2))))), + ?assert(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return2))))), {ok, Return3} = request_api(get, api_path(["alarms/deactivated"]), auth_header_()), - ?assert(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return3))))), - ?assertNot(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return3))))), + ?assert(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return3))))), + ?assertNot(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return3))))), emqx_alarm:deactivate(alarm2), {ok, Return4} = request_api(get, api_path(["alarms/deactivated"]), auth_header_()), - ?assert(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return4))))), - ?assert(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return4))))), + ?assert(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return4))))), + ?assert(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return4))))), {ok, _} = request_api(delete, api_path(["alarms/deactivated"]), auth_header_()), {ok, Return5} = request_api(get, api_path(["alarms/deactivated"]), auth_header_()), - ?assertNot(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return5))))), - ?assertNot(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return5))))). + ?assertNot(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return5))))), + ?assertNot(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, + lists:nth(1, get(<<"data">>, Return5))))). t_apps(_) -> AppId = <<"123456">>, @@ -153,7 +165,8 @@ t_banned(_) -> [Banned] = get(<<"data">>, Result), ?assertEqual(Who, maps:get(<<"who">>, Banned)), - {ok, _} = request_api(delete, api_path(["banned", "clientid", binary_to_list(Who)]), auth_header_()), + {ok, _} = request_api(delete, api_path(["banned", "clientid", binary_to_list(Who)]), + auth_header_()), {ok, Result2} = request_api(get, api_path(["banned"]), auth_header_()), ?assertEqual([], get(<<"data">>, Result2)). @@ -205,40 +218,50 @@ t_clients(_) -> meck:new(emqx_mgmt, [passthrough, no_history]), meck:expect(emqx_mgmt, kickout_client, 1, fun(_) -> {error, undefined} end), - {ok, MeckRet1} = request_api(delete, api_path(["clients", binary_to_list(ClientId1)]), auth_header_()), + {ok, MeckRet1} = request_api(delete, api_path(["clients", binary_to_list(ClientId1)]), + auth_header_()), ?assertEqual(?ERROR1, get(<<"code">>, MeckRet1)), meck:expect(emqx_mgmt, clean_acl_cache, 1, fun(_) -> {error, undefined} end), - {ok, MeckRet2} = request_api(delete, api_path(["clients", binary_to_list(ClientId1), "acl_cache"]), auth_header_()), + {ok, MeckRet2} = request_api(delete, + api_path(["clients", binary_to_list(ClientId1), "acl_cache"]), auth_header_()), ?assertEqual(?ERROR1, get(<<"code">>, MeckRet2)), meck:expect(emqx_mgmt, list_acl_cache, 1, fun(_) -> {error, undefined} end), - {ok, MeckRet3} = request_api(get, api_path(["clients", binary_to_list(ClientId2), "acl_cache"]), auth_header_()), + {ok, MeckRet3} = request_api(get, + api_path(["clients", binary_to_list(ClientId2), "acl_cache"]), auth_header_()), ?assertEqual(?ERROR1, get(<<"code">>, MeckRet3)), meck:unload(emqx_mgmt), - {ok, Ok} = request_api(delete, api_path(["clients", binary_to_list(ClientId1)]), auth_header_()), + {ok, Ok} = request_api(delete, + api_path(["clients", binary_to_list(ClientId1)]), auth_header_()), ?assertEqual(?SUCCESS, get(<<"code">>, Ok)), timer:sleep(300), - {ok, Ok1} = request_api(delete, api_path(["clients", binary_to_list(ClientId1)]), auth_header_()), + {ok, Ok1} = request_api(delete, + api_path(["clients", binary_to_list(ClientId1)]), auth_header_()), ?assertEqual(?SUCCESS, get(<<"code">>, Ok1)), - {ok, Clients6} = request_api(get, api_path(["clients"]), "_limit=100&_page=1", auth_header_()), + {ok, Clients6} = request_api(get, + api_path(["clients"]), "_limit=100&_page=1", auth_header_()), ?assertEqual(1, maps:get(<<"count">>, get(<<"meta">>, Clients6))), - {ok, NotFound1} = request_api(get, api_path(["clients", binary_to_list(ClientId1), "acl_cache"]), auth_header_()), + {ok, NotFound1} = request_api(get, + api_path(["clients", binary_to_list(ClientId1), "acl_cache"]), auth_header_()), ?assertEqual(?ERROR12, get(<<"code">>, NotFound1)), - {ok, NotFound2} = request_api(delete, api_path(["clients", binary_to_list(ClientId1), "acl_cache"]), auth_header_()), + {ok, NotFound2} = request_api(delete, + api_path(["clients", binary_to_list(ClientId1), "acl_cache"]), auth_header_()), ?assertEqual(?ERROR12, get(<<"code">>, NotFound2)), - {ok, EmptyAclCache} = request_api(get, api_path(["clients", binary_to_list(ClientId2), "acl_cache"]), auth_header_()), + {ok, EmptyAclCache} = request_api(get, + api_path(["clients", binary_to_list(ClientId2), "acl_cache"]), auth_header_()), ?assertEqual(0, length(get(<<"data">>, EmptyAclCache))), - {ok, Ok1} = request_api(delete, api_path(["clients", binary_to_list(ClientId2), "acl_cache"]), auth_header_()), + {ok, Ok1} = request_api(delete, + api_path(["clients", binary_to_list(ClientId2), "acl_cache"]), auth_header_()), ?assertEqual(?SUCCESS, get(<<"code">>, Ok1)). receive_exit(0) -> @@ -257,7 +280,8 @@ receive_exit(Count) -> t_listeners(_) -> {ok, _} = request_api(get, api_path(["listeners"]), auth_header_()), - {ok, _} = request_api(get, api_path(["nodes", atom_to_list(node()), "listeners"]), auth_header_()), + {ok, _} = request_api(get, + api_path(["nodes", atom_to_list(node()), "listeners"]), auth_header_()), meck:new(emqx_mgmt, [passthrough, no_history]), meck:expect(emqx_mgmt, list_listeners, 0, fun() -> [{node(), {error, undefined}}] end), {ok, Return} = request_api(get, api_path(["listeners"]), auth_header_()), @@ -268,10 +292,12 @@ t_listeners(_) -> t_metrics(_) -> {ok, _} = request_api(get, api_path(["metrics"]), auth_header_()), - {ok, _} = request_api(get, api_path(["nodes", atom_to_list(node()), "metrics"]), auth_header_()), + {ok, _} = request_api(get, + api_path(["nodes", atom_to_list(node()), "metrics"]), auth_header_()), meck:new(emqx_mgmt, [passthrough, no_history]), meck:expect(emqx_mgmt, get_metrics, 1, fun(_) -> {error, undefined} end), - {ok, "{\"message\":\"undefined\"}"} = request_api(get, api_path(["nodes", atom_to_list(node()), "metrics"]), auth_header_()), + {ok, "{\"message\":\"undefined\"}"} = + request_api(get, api_path(["nodes", atom_to_list(node()), "metrics"]), auth_header_()), meck:unload(emqx_mgmt). t_nodes(_) -> @@ -348,7 +374,8 @@ t_acl_cache(_) -> {ok, _} = emqtt:connect(C1), {ok, _, _} = emqtt:subscribe(C1, Topic, 2), %% get acl cache, should not be empty - {ok, Result} = request_api(get, api_path(["clients", binary_to_list(ClientId), "acl_cache"]), [], auth_header_()), + {ok, Result} = request_api(get, + api_path(["clients", binary_to_list(ClientId), "acl_cache"]), [], auth_header_()), #{<<"code">> := 0, <<"data">> := Caches} = jiffy:decode(list_to_binary(Result), [return_maps]), ?assert(length(Caches) > 0), ?assertMatch(#{<<"access">> := <<"subscribe">>, @@ -356,11 +383,14 @@ t_acl_cache(_) -> <<"result">> := <<"allow">>, <<"updated_time">> := _}, hd(Caches)), %% clear acl cache - {ok, Result2} = request_api(delete, api_path(["clients", binary_to_list(ClientId), "acl_cache"]), [], auth_header_()), + {ok, Result2} = request_api(delete, + api_path(["clients", binary_to_list(ClientId), "acl_cache"]), [], auth_header_()), ?assertMatch(#{<<"code">> := 0}, jiffy:decode(list_to_binary(Result2), [return_maps])), %% get acl cache again, after the acl cache is cleared - {ok, Result3} = request_api(get, api_path(["clients", binary_to_list(ClientId), "acl_cache"]), [], auth_header_()), - #{<<"code">> := 0, <<"data">> := Caches3} = jiffy:decode(list_to_binary(Result3), [return_maps]), + {ok, Result3} = request_api(get, + api_path(["clients", binary_to_list(ClientId), "acl_cache"]), [], auth_header_()), + #{<<"code">> := 0, <<"data">> := Caches3} + = jiffy:decode(list_to_binary(Result3), [return_maps]), ?assertEqual(0, length(Caches3)), ok = emqtt:disconnect(C1). @@ -482,12 +512,15 @@ t_pubsub(_) -> Topic_list = [<<"mytopic1">>, <<"mytopic2">>], [ {ok, _, [2]} = emqtt:subscribe(C1, Topics, 2) || Topics <- Topic_list], - Body1 = [ #{<<"clientid">> => ClientId, <<"topic">> => Topics, <<"qos">> => 2} || Topics <- Topic_list], + Body1 = [ #{<<"clientid">> => ClientId, + <<"topic">> => Topics, <<"qos">> => 2} || Topics <- Topic_list], {ok, Data1} = request_api(post, api_path(["mqtt/subscribe_batch"]), [], auth_header_(), Body1), loop(maps:get(<<"data">>, jiffy:decode(list_to_binary(Data1), [return_maps]))), %% tests publish_batch - Body2 = [ #{<<"clientid">> => ClientId, <<"topic">> => Topics, <<"qos">> => 2, <<"retain">> => <<"false">>, <<"payload">> => #{body => "hello world"}} || Topics <- Topic_list ], + Body2 = [ #{<<"clientid">> => ClientId, <<"topic">> => Topics, <<"qos">> => 2, + <<"retain">> => <<"false">>, <<"payload">> => #{body => "hello world"}} + || Topics <- Topic_list ], {ok, Data2} = request_api(post, api_path(["mqtt/publish_batch"]), [], auth_header_(), Body2), loop(maps:get(<<"data">>, jiffy:decode(list_to_binary(Data2), [return_maps]))), [ ?assert(receive @@ -499,7 +532,8 @@ t_pubsub(_) -> %% tests unsubscribe_batch Body3 = [#{<<"clientid">> => ClientId, <<"topic">> => Topics} || Topics <- Topic_list], - {ok, Data3} = request_api(post, api_path(["mqtt/unsubscribe_batch"]), [], auth_header_(), Body3), + {ok, Data3} = request_api(post, + api_path(["mqtt/unsubscribe_batch"]), [], auth_header_(), Body3), loop(maps:get(<<"data">>, jiffy:decode(list_to_binary(Data3), [return_maps]))), ok = emqtt:disconnect(C1), @@ -523,7 +557,8 @@ t_routes_and_subscriptions(_) -> ?assertEqual([], get(<<"data">>, NonRoute)), {ok, NonSubscription} = request_api(get, api_path(["subscriptions"]), auth_header_()), ?assertEqual([], get(<<"data">>, NonSubscription)), - {ok, NonSubscription1} = request_api(get, api_path(["nodes", atom_to_list(node()), "subscriptions"]), auth_header_()), + {ok, NonSubscription1} = request_api(get, + api_path(["nodes", atom_to_list(node()), "subscriptions"]), auth_header_()), ?assertEqual([], get(<<"data">>, NonSubscription1)), {ok, NonSubscription2} = request_api(get, api_path(["subscriptions", binary_to_list(ClientId)]), @@ -552,11 +587,14 @@ t_routes_and_subscriptions(_) -> ?assertEqual(Topic, maps:get(<<"topic">>, Subscription)), ?assertEqual(ClientId, maps:get(<<"clientid">>, Subscription)), - {ok, Result3} = request_api(get, api_path(["nodes", atom_to_list(node()), "subscriptions"]), auth_header_()), + {ok, Result3} = request_api(get, + api_path(["nodes", atom_to_list(node()), "subscriptions"]), auth_header_()), - {ok, Result4} = request_api(get, api_path(["subscriptions", binary_to_list(ClientId)]), auth_header_()), + {ok, Result4} = request_api(get, + api_path(["subscriptions", binary_to_list(ClientId)]), auth_header_()), [Subscription] = get(<<"data">>, Result4), - {ok, Result4} = request_api(get, api_path(["nodes", atom_to_list(node()), "subscriptions", binary_to_list(ClientId)]) + {ok, Result4} = request_api(get, + api_path(["nodes", atom_to_list(node()), "subscriptions", binary_to_list(ClientId)]) , auth_header_()), ok = emqtt:disconnect(C1). @@ -566,7 +604,8 @@ t_stats(_) -> {ok, _} = request_api(get, api_path(["nodes", atom_to_list(node()), "stats"]), auth_header_()), meck:new(emqx_mgmt, [passthrough, no_history]), meck:expect(emqx_mgmt, get_stats, 1, fun(_) -> {error, undefined} end), - {ok, Return} = request_api(get, api_path(["nodes", atom_to_list(node()), "stats"]), auth_header_()), + {ok, Return} = request_api(get, + api_path(["nodes", atom_to_list(node()), "stats"]), auth_header_()), ?assertEqual(<<"undefined">>, get(<<"message">>, Return)), meck:unload(emqx_mgmt). @@ -578,10 +617,15 @@ t_data(_) -> {ok, Data} = request_api(post, api_path(["data","export"]), [], auth_header_(), [#{}]), #{<<"filename">> := Filename, <<"node">> := Node} = emqx_ct_http:get_http_data(Data), {ok, DataList} = request_api(get, api_path(["data","export"]), auth_header_()), - ?assertEqual(true, lists:member(emqx_ct_http:get_http_data(Data), emqx_ct_http:get_http_data(DataList))), + ?assertEqual(true, + lists:member(emqx_ct_http:get_http_data(Data), emqx_ct_http:get_http_data(DataList))), - ?assertMatch({ok, _}, request_api(post, api_path(["data","import"]), [], auth_header_(), #{<<"filename">> => Filename, <<"node">> => Node})), - ?assertMatch({ok, _}, request_api(post, api_path(["data","import"]), [], auth_header_(), #{<<"filename">> => Filename})), + ?assertMatch({ok, _}, request_api(post, + api_path(["data","import"]), [], auth_header_(), + #{<<"filename">> => Filename, <<"node">> => Node})), + ?assertMatch({ok, _}, + request_api(post, api_path(["data","import"]), [], auth_header_(), + #{<<"filename">> => Filename})), application:stop(emqx_rule_engine), application:stop(emqx_dahboard), ok. @@ -596,9 +640,26 @@ t_data_import_content(_) -> Dir = emqx:get_env(data_dir), {ok, Bin} = file:read_file(filename:join(Dir, Filename)), Content = emqx_json:decode(Bin), - ?assertMatch({ok, "{\"code\":0}"}, request_api(post, api_path(["data","import"]), [], auth_header_(), Content)), + ?assertMatch({ok, "{\"code\":0}"}, + request_api(post, api_path(["data","import"]), [], auth_header_(), Content)), application:stop(emqx_rule_engine), - application:stop(emqx_dahboard). + application:stop(emqx_dashboard). + +t_keepalive(_Config) -> + application:ensure_all_started(emqx_dashboard), + Username = "user_keepalive", + ClientId = "client_keepalive", + AuthHeader = auth_header_(), + Path = api_path(["clients", ClientId, "keepalive"]), + {ok, NotFound} = request_api(put, Path, "interval=5", AuthHeader, [#{}]), + ?assertEqual("{\"message\":\"not_found\",\"code\":112}", NotFound), + {ok, C1} = emqtt:start_link(#{username => Username, clientid => ClientId}), + {ok, _} = emqtt:connect(C1), + {ok, Ok} = request_api(put, Path, "interval=5", AuthHeader, [#{}]), + ?assertEqual("{\"code\":0}", Ok), + emqtt:disconnect(C1), + application:stop(emqx_dashboard), + ok. request_api(Method, Url, Auth) -> request_api(Method, Url, [], Auth, []). diff --git a/apps/emqx_retainer/src/emqx_retainer.app.src b/apps/emqx_retainer/src/emqx_retainer.app.src index c3ffb9f90..a423fb9b7 100644 --- a/apps/emqx_retainer/src/emqx_retainer.app.src +++ b/apps/emqx_retainer/src/emqx_retainer.app.src @@ -1,6 +1,6 @@ {application, emqx_retainer, [{description, "EMQ X Retainer"}, - {vsn, "4.3.1"}, % strict semver, bump manually! + {vsn, "4.4.0"}, % strict semver, bump manually! {modules, []}, {registered, [emqx_retainer_sup]}, {applications, [kernel,stdlib]}, diff --git a/src/emqx_channel.erl b/src/emqx_channel.erl index 5f164d99e..4c49a1cb3 100644 --- a/src/emqx_channel.erl +++ b/src/emqx_channel.erl @@ -102,7 +102,7 @@ -type(reply() :: {outgoing, emqx_types:packet()} | {outgoing, [emqx_types:packet()]} - | {event, conn_state()|updated} + | {event, conn_state() | updated} | {close, Reason :: atom()}). -type(replies() :: emqx_types:packet() | reply() | [reply()]). @@ -131,7 +131,7 @@ info(Channel) -> maps:from_list(info(?INFO_KEYS, Channel)). --spec(info(list(atom())|atom(), channel()) -> term()). +-spec(info(list(atom()) | atom(), channel()) -> term()). info(Keys, Channel) when is_list(Keys) -> [{Key, info(Key, Channel)} || Key <- Keys]; info(conninfo, #channel{conninfo = ConnInfo}) -> @@ -619,7 +619,7 @@ ensure_quota(PubRes, Channel = #channel{quota = Limiter}) -> -compile({inline, [puback_reason_code/1]}). puback_reason_code([]) -> ?RC_NO_MATCHING_SUBSCRIBERS; -puback_reason_code([_|_]) -> ?RC_SUCCESS. +puback_reason_code([_ | _]) -> ?RC_SUCCESS. -compile({inline, [after_message_acked/3]}). after_message_acked(ClientInfo, Msg, PubAckProps) -> @@ -638,7 +638,7 @@ process_subscribe(TopicFilters, SubProps, Channel) -> process_subscribe([], _SubProps, Channel, Acc) -> {lists:reverse(Acc), Channel}; -process_subscribe([Topic = {TopicFilter, SubOpts}|More], SubProps, Channel, Acc) -> +process_subscribe([Topic = {TopicFilter, SubOpts} | More], SubProps, Channel, Acc) -> case check_sub_caps(TopicFilter, SubOpts, Channel) of ok -> {ReasonCode, NChannel} = do_subscribe(TopicFilter, @@ -676,9 +676,9 @@ process_unsubscribe(TopicFilters, UnSubProps, Channel) -> process_unsubscribe([], _UnSubProps, Channel, Acc) -> {lists:reverse(Acc), Channel}; -process_unsubscribe([{TopicFilter, SubOpts}|More], UnSubProps, Channel, Acc) -> +process_unsubscribe([{TopicFilter, SubOpts} | More], UnSubProps, Channel, Acc) -> {RC, NChannel} = do_unsubscribe(TopicFilter, SubOpts#{unsub_props => UnSubProps}, Channel), - process_unsubscribe(More, UnSubProps, NChannel, [RC|Acc]). + process_unsubscribe(More, UnSubProps, NChannel, [RC | Acc]). do_unsubscribe(TopicFilter, SubOpts, Channel = #channel{clientinfo = ClientInfo = #{mountpoint := MountPoint}, @@ -938,6 +938,17 @@ handle_call({quota, Policy}, Channel) -> Quota = emqx_limiter:init(Zone, Policy), reply(ok, Channel#channel{quota = Quota}); +handle_call({keepalive, Interval}, Channel = #channel{keepalive = KeepAlive, + conninfo = ConnInfo}) -> + ClientId = info(clientid, Channel), + NKeepalive = emqx_keepalive:set(interval, Interval * 1000, KeepAlive), + NConnInfo = maps:put(keepalive, Interval, ConnInfo), + NChannel = Channel#channel{keepalive = NKeepalive, conninfo = NConnInfo}, + SockInfo = maps:get(sockinfo, emqx_cm:get_chan_info(ClientId), #{}), + ChanInfo1 = info(NChannel), + emqx_cm:set_chan_info(ClientId, ChanInfo1#{sockinfo => SockInfo}), + reply(ok, reset_timer(alive_timer, NChannel)); + handle_call(Req, Channel) -> ?LOG(error, "Unexpected call: ~p", [Req]), reply(ignored, Channel). diff --git a/src/emqx_keepalive.erl b/src/emqx_keepalive.erl index 8fba00f50..216999acc 100644 --- a/src/emqx_keepalive.erl +++ b/src/emqx_keepalive.erl @@ -20,9 +20,11 @@ , info/1 , info/2 , check/2 + , set/3 ]). -export_type([keepalive/0]). +-elvis([{elvis_style, no_if_expression, disable}]). -record(keepalive, { interval :: pos_integer(), @@ -49,7 +51,7 @@ info(#keepalive{interval = Interval, repeat => Repeat }. --spec(info(interval|statval|repeat, keepalive()) +-spec(info(interval | statval | repeat, keepalive()) -> non_neg_integer()). info(interval, #keepalive{interval = Interval}) -> Interval; @@ -71,3 +73,7 @@ check(NewVal, KeepAlive = #keepalive{statval = OldVal, true -> {error, timeout} end. +%% @doc Update keepalive's interval +-spec(set(interval, non_neg_integer(), keepalive()) -> keepalive()). +set(interval, Interval, KeepAlive) -> + KeepAlive#keepalive{interval = Interval}.