chore(emqx_modules): improve emqx_delayed coverage

apps/emqx_modules/src/emqx_delayed.erl

@@ -21,6 +21,7 @@
 -include_lib("emqx/include/emqx.hrl").
 -include_lib("emqx/include/types.hrl").
 -include_lib("emqx/include/logger.hrl").
+-include_lib("snabbkaffe/include/snabbkaffe.hrl").
 
 %% Mnesia bootstrap
 -export([mnesia/1]).
@@ -275,17 +276,13 @@ init([Opts]) ->
 handle_call({set_max_delayed_messages, Max}, _From, State) ->
     {reply, ok, State#{max_delayed_messages => Max}};
 handle_call(
-    {store, DelayedMsg = #delayed_message{key = Key}},
+    {store, DelayedMsg = #delayed_message{key = Key}}, _From, State = #{max_delayed_messages := 0}
-    _From,
-    State = #{max_delayed_messages := 0}
 ) ->
     ok = mria:dirty_write(?TAB, DelayedMsg),
     emqx_metrics:inc('messages.delayed'),
     {reply, ok, ensure_publish_timer(Key, State)};
 handle_call(
-    {store, DelayedMsg = #delayed_message{key = Key}},
+    {store, DelayedMsg = #delayed_message{key = Key}}, _From, State = #{max_delayed_messages := Max}
-    _From,
-    State = #{max_delayed_messages := Max}
 ) ->
     Size = mnesia:table_info(?TAB, size),
     case Size >= Max of
@@ -303,11 +300,11 @@ handle_call(disable, _From, State) ->
     emqx_hooks:del('message.publish', {?MODULE, on_message_publish}),
     {reply, ok, State};
 handle_call(Req, _From, State) ->
-    ?SLOG(error, #{msg => "unexpected_call", call => Req}),
+    ?tp(error, emqx_delayed_unexpected_call, #{call => Req}),
     {reply, ignored, State}.
 
 handle_cast(Msg, State) ->
-    ?SLOG(error, #{msg => "unexpected_cast", cast => Msg}),
+    ?tp(error, emqx_delayed_unexpected_cast, #{cast => Msg}),
     {noreply, State}.
 
 %% Do Publish...
@@ -320,7 +317,7 @@ handle_info(stats, State = #{stats_fun := StatsFun}) ->
     StatsFun(delayed_count()),
     {noreply, State#{stats_timer := StatsTimer}, hibernate};
 handle_info(Info, State) ->
-    ?SLOG(error, #{msg => "unexpected_info", info => Info}),
+    ?tp(error, emqx_delayed_unexpected_info, #{info => Info}),
     {noreply, State}.
 
 terminate(_Reason, #{publish_timer := PublishTimer, stats_timer := StatsTimer}) ->

apps/emqx_modules/src/proto/emqx_delayed_proto_v1.erl

@@ -29,9 +29,9 @@ introduced_in() ->
     "5.0.0".
 
 -spec get_delayed_message(node(), binary()) -> emqx_delayed:with_id_return(map()) | emqx_rpc:badrpc().
-get_delayed_message(Node, HexId) ->
+get_delayed_message(Node, Id) ->
-    rpc:call(Node, emqx_delayed, get_delayed_message, [HexId]).
+    rpc:call(Node, emqx_delayed, get_delayed_message, [Id]).
 
 -spec delete_delayed_message(node(), binary()) -> emqx_delayed:with_id_return() | emqx_rpc:badrpc().
-delete_delayed_message(Node, HexId) ->
+delete_delayed_message(Node, Id) ->
-    rpc:call(Node, emqx_delayed, delete_delayed_message, [HexId]).
+    rpc:call(Node, emqx_delayed, delete_delayed_message, [Id]).

apps/emqx_modules/test/emqx_delayed_SUITE.erl

@@ -43,6 +43,16 @@ init_per_suite(Config) ->
 end_per_suite(_) ->
     emqx_common_test_helpers:stop_apps([emqx_modules]).
 
+init_per_testcase(t_load_case, Config) ->
+    Config;
+init_per_testcase(_Case, Config) ->
+    {atomic, ok} = mria:clear_table(emqx_delayed),
+    ok = emqx_delayed:enable(),
+    Config.
+
+end_per_testcase(_Case, _Config) ->
+    ok = emqx_delayed:disable().
+
 %%--------------------------------------------------------------------
 %% Test cases
 %%--------------------------------------------------------------------
@@ -57,7 +67,6 @@ t_load_case(_) ->
     ok.
 
 t_delayed_message(_) ->
-    ok = emqx_delayed:enable(),
     DelayedMsg = emqx_message:make(?MODULE, 1, <<"$delayed/1/publish">>, <<"delayed_m">>),
     ?assertEqual(
         {stop, DelayedMsg#message{topic = <<"publish">>, headers = #{allow_publish => false}}},
@@ -67,11 +76,94 @@ t_delayed_message(_) ->
     Msg = emqx_message:make(?MODULE, 1, <<"no_delayed_msg">>, <<"no_delayed">>),
     ?assertEqual({ok, Msg}, on_message_publish(Msg)),
 
-    [Key] = mnesia:dirty_all_keys(emqx_delayed),
+    [#delayed_message{msg = #message{payload = Payload}}] = ets:tab2list(emqx_delayed),
-    [#delayed_message{msg = #message{payload = Payload}}] = mnesia:dirty_read({emqx_delayed, Key}),
     ?assertEqual(<<"delayed_m">>, Payload),
-    timer:sleep(5000),
+    ct:sleep(2000),
 
     EmptyKey = mnesia:dirty_all_keys(emqx_delayed),
-    ?assertEqual([], EmptyKey),
+    ?assertEqual([], EmptyKey).
-    ok = emqx_delayed:disable().
+
+t_delayed_message_abs_time(_) ->
+    Ts0 = integer_to_binary(erlang:system_time(second) + 1),
+    DelayedMsg0 = emqx_message:make(
+        ?MODULE, 1, <<"$delayed/", Ts0/binary, "/publish">>, <<"delayed_abs">>
+    ),
+    _ = on_message_publish(DelayedMsg0),
+
+    ?assertMatch(
+        [#delayed_message{msg = #message{payload = <<"delayed_abs">>}}],
+        ets:tab2list(emqx_delayed)
+    ),
+
+    ct:sleep(2000),
+
+    ?assertMatch(
+        [],
+        ets:tab2list(emqx_delayed)
+    ),
+
+    Ts1 = integer_to_binary(erlang:system_time(second) + 10000000),
+    DelayedMsg1 = emqx_message:make(
+        ?MODULE, 1, <<"$delayed/", Ts1/binary, "/publish">>, <<"delayed_abs">>
+    ),
+
+    ?assertError(
+        invalid_delayed_timestamp,
+        on_message_publish(DelayedMsg1)
+    ).
+
+t_list(_) ->
+    Ts0 = integer_to_binary(erlang:system_time(second) + 1),
+    DelayedMsg0 = emqx_message:make(
+        ?MODULE, 1, <<"$delayed/", Ts0/binary, "/publish">>, <<"delayed_abs">>
+    ),
+    _ = on_message_publish(DelayedMsg0),
+
+    ?assertMatch(
+        #{data := [#{topic := <<"publish">>}]},
+        emqx_delayed:list(#{})
+    ).
+
+t_max(_) ->
+    emqx_delayed:set_max_delayed_messages(1),
+
+    DelayedMsg0 = emqx_message:make(?MODULE, 1, <<"$delayed/10/t0">>, <<"delayed0">>),
+    DelayedMsg1 = emqx_message:make(?MODULE, 1, <<"$delayed/10/t1">>, <<"delayed1">>),
+    _ = on_message_publish(DelayedMsg0),
+    _ = on_message_publish(DelayedMsg1),
+
+    ?assertMatch(
+        #{data := [#{topic := <<"t0">>}]},
+        emqx_delayed:list(#{})
+    ).
+
+t_cluster(_) ->
+    DelayedMsg = emqx_message:make(?MODULE, 1, <<"$delayed/1/publish">>, <<"delayed">>),
+    Id = emqx_message:id(DelayedMsg),
+    _ = on_message_publish(DelayedMsg),
+
+    ?assertMatch(
+        {ok, _},
+        emqx_delayed_proto_v1:get_delayed_message(node(), Id)
+    ),
+
+    ?assertEqual(
+        emqx_delayed:get_delayed_message(Id),
+        emqx_delayed_proto_v1:get_delayed_message(node(), Id)
+    ),
+
+    ok = emqx_delayed_proto_v1:delete_delayed_message(node(), Id),
+
+    ?assertMatch(
+        {error, _},
+        emqx_delayed:get_delayed_message(Id)
+    ).
+
+t_unknown_messages(_) ->
+    OldPid = whereis(emqx_delayed),
+    OldPid ! unknown,
+    ok = gen_server:cast(OldPid, unknown),
+    ?assertEqual(
+        ignored,
+        gen_server:call(OldPid, unknown)
+    ).

apps/emqx_modules/test/emqx_delayed_api_SUITE.erl

@@ -20,11 +20,11 @@
 
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
--include_lib("emqx/include/emqx.hrl").
 
--define(BASE_CONF, #{<<"dealyed">> => <<"true">>,
+-define(BASE_CONF, #{
-                     <<"max_delayed_messages">> => <<"0">>
+    <<"dealyed">> => <<"true">>,
-                    }).
+    <<"max_delayed_messages">> => <<"0">>
+}).
 
 -import(emqx_dashboard_api_test_helpers, [request/2, request/3, uri/1]).
 
@@ -75,20 +75,22 @@ t_status(_Config) ->
     ?assertMatch(#{enable := true, max_delayed_messages := 12}, decode_json(R2)),
 
     ?assertMatch(
-       {ok, 200, _},
+        {ok, 200, _},
-       request(
+        request(
-         put,
+            put,
-         Path,
+            Path,
-         #{enable => true}
+            #{enable => true}
-        )),
+        )
+    ),
 
     ?assertMatch(
-       {ok, 400, _},
+        {ok, 400, _},
-       request(
+        request(
-         put,
+            put,
-         Path,
+            Path,
-         #{enable => true, max_delayed_messages => -5}
+            #{enable => true, max_delayed_messages => -5}
-        )),
+        )
+    ),
 
     {ok, 200, ConfJson} = request(get, Path),
     ReturnConf = decode_json(ConfJson),
@@ -201,11 +203,11 @@ t_large_payload(_) ->
     timer:sleep(500),
     Topic = <<"$delayed/123/msgs">>,
     emqtt:publish(
-      C1,
+        C1,
-      Topic,
+        Topic,
-      iolist_to_binary([<<"x">> || _ <- lists:seq(1, 5000)]),
+        iolist_to_binary([<<"x">> || _ <- lists:seq(1, 5000)]),
-      [{qos, 0}, {retain, true}]
+        [{qos, 0}, {retain, true}]
-     ),
+    ),
 
     timer:sleep(500),