diff --git a/apps/emqx/include/emqx.hrl b/apps/emqx/include/emqx.hrl index 67744306e..d148e01a3 100644 --- a/apps/emqx/include/emqx.hrl +++ b/apps/emqx/include/emqx.hrl @@ -25,6 +25,7 @@ -define(COMMON_SHARD, emqx_common_shard). -define(SHARED_SUB_SHARD, emqx_shared_sub_shard). +-define(MOD_DELAYED_SHARD, emqx_delayed_shard). %%-------------------------------------------------------------------- %% Banner diff --git a/apps/emqx/src/emqx_app.erl b/apps/emqx/src/emqx_app.erl index 06bebe465..60f0fc40d 100644 --- a/apps/emqx/src/emqx_app.erl +++ b/apps/emqx/src/emqx_app.erl @@ -32,6 +32,7 @@ , ?COMMON_SHARD , ?SHARED_SUB_SHARD , ?RULE_ENGINE_SHARD + , ?MOD_DELAYED_SHARD ]). -include("emqx_release.hrl"). diff --git a/apps/emqx_modules/src/emqx_mod_delayed.erl b/apps/emqx_modules/src/emqx_mod_delayed.erl index ac5be58b2..925952078 100644 --- a/apps/emqx_modules/src/emqx_mod_delayed.erl +++ b/apps/emqx_modules/src/emqx_mod_delayed.erl @@ -58,6 +58,8 @@ -define(SERVER, ?MODULE). -define(MAX_INTERVAL, 4294967). +-rlog_shard({?MOD_DELAYED_SHARD, ?TAB}). + %%-------------------------------------------------------------------- %% Mnesia bootstrap %%-------------------------------------------------------------------- @@ -137,7 +139,7 @@ init([]) -> ensure_publish_timer(#{timer => undefined, publish_at => 0}))}. handle_call({store, DelayedMsg = #delayed_message{key = Key}}, _From, State) -> - ok = mnesia:dirty_write(?TAB, DelayedMsg), + ok = ekka_mnesia:dirty_write(?TAB, DelayedMsg), emqx_metrics:inc('messages.delayed'), {reply, ok, ensure_publish_timer(Key, State)}; @@ -152,7 +154,7 @@ handle_cast(Msg, State) -> %% Do Publish... handle_info({timeout, TRef, do_publish}, State = #{timer := TRef}) -> DeletedKeys = do_publish(mnesia:dirty_first(?TAB), os:system_time(seconds)), - lists:foreach(fun(Key) -> mnesia:dirty_delete(?TAB, Key) end, DeletedKeys), + lists:foreach(fun(Key) -> ekka_mnesia:dirty_delete(?TAB, Key) end, DeletedKeys), {noreply, ensure_publish_timer(State#{timer := undefined, publish_at := 0})}; handle_info(stats, State = #{stats_fun := StatsFun}) -> @@ -222,4 +224,3 @@ do_publish(Key = {Ts, _Id}, Now, Acc) when Ts =< Now -> -spec(delayed_count() -> non_neg_integer()). delayed_count() -> mnesia:table_info(?TAB, size). -