Set keepalive via http api (#6143)

* feat: set keepalive over http api

* chore: elvis warning

* fix: bump retainer to 4.4.0

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
+    has_any_ok(Results).
-        true  -> ok;
-        false -> lists:last(Results)
-    end.
 
 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
+    has_any_ok(Results).
-        true  -> ok;
-        false -> lists:last(Results)
-    end.
 
 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.

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(QUERY_FUN, {?MODULE, query}).
--define(format_fun, {?MODULE, format_channel_info}).
+-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
 %%--------------------------------------------------------------------

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(<<"alarm1">>, maps:get(<<"alarms">>,
-    ?assert(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return1))))),
+        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(<<"alarm1">>, maps:get(<<"alarms">>,
-    ?assert(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return2))))),
+        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))))),
+    ?assert(lookup_alarm(<<"alarm1">>, maps:get(<<"alarms">>,
-    ?assertNot(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return3))))),
+        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(<<"alarm1">>, maps:get(<<"alarms">>,
-    ?assert(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return4))))),
+        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(<<"alarm1">>, maps:get(<<"alarms">>,
-    ?assertNot(lookup_alarm(<<"alarm2">>, maps:get(<<"alarms">>, lists:nth(1, get(<<"data">>, Return5))))).
+        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_()),
+    {ok, Result3} = request_api(get,
-    #{<<"code">> := 0, <<"data">> := Caches3} = jiffy:decode(list_to_binary(Result3), [return_maps]),
+        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,
-    ?assertMatch({ok, _}, request_api(post, api_path(["data","import"]), [], auth_header_(), #{<<"filename">> => Filename})),
+        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, []).

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]},

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).

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}.