chore(mria): ekka_mnesia:create_table -> mria:create_table

apps/emqx/src/emqx_alarm.erl

@@ -95,21 +95,18 @@
 %%--------------------------------------------------------------------
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?ACTIVATED_ALARM,
+    ok = mria:create_table(?ACTIVATED_ALARM,
              [{type, set},
-              {disc_copies, [node()]},
+              {storage, disc_copies},
               {local_content, true},
               {record_name, activated_alarm},
               {attributes, record_info(fields, activated_alarm)}]),
-    ok = ekka_mnesia:create_table(?DEACTIVATED_ALARM,
+    ok = mria:create_table(?DEACTIVATED_ALARM,
              [{type, ordered_set},
-              {disc_copies, [node()]},
+              {storage, disc_copies},
               {local_content, true},
               {record_name, deactivated_alarm},
-              {attributes, record_info(fields, deactivated_alarm)}]);
+              {attributes, record_info(fields, deactivated_alarm)}]).
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?ACTIVATED_ALARM, disc_copies),
-    ok = ekka_mnesia:copy_table(?DEACTIVATED_ALARM, disc_copies).
 
 %%--------------------------------------------------------------------
 %% API

apps/emqx/src/emqx_banned.erl

@@ -58,16 +58,13 @@
 %%--------------------------------------------------------------------
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?BANNED_TAB, [
+    ok = mria:create_table(?BANNED_TAB, [
                 {type, set},
                 {rlog_shard, ?COMMON_SHARD},
-                {disc_copies, [node()]},
+                {storage, disc_copies},
                 {record_name, banned},
                 {attributes, record_info(fields, banned)},
-                {storage_properties, [{ets, [{read_concurrency, true}]}]}]);
+                {storage_properties, [{ets, [{read_concurrency, true}]}]}]).
-
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?BANNED_TAB, disc_copies).
 
 %% @doc Start the banned server.
 -spec(start_link() -> startlink_ret()).

apps/emqx/src/emqx_cm_registry.erl

@@ -100,15 +100,14 @@ record(ClientId, ChanPid) ->
 %%--------------------------------------------------------------------
 
 init([]) ->
-    ok = ekka_mnesia:create_table(?TAB, [
+    ok = mria:create_table(?TAB, [
                 {type, bag},
                 {rlog_shard, ?CM_SHARD},
-                {ram_copies, [node()]},
+                {storage, ram_copies},
                 {record_name, channel},
                 {attributes, record_info(fields, channel)},
                 {storage_properties, [{ets, [{read_concurrency, true},
                                              {write_concurrency, true}]}]}]),
-    ok = ekka_mnesia:copy_table(?TAB, ram_copies),
     ok = ekka_rlog:wait_for_shards([?CM_SHARD], infinity),
     ok = ekka:monitor(membership),
     {ok, #{}}.

apps/emqx/src/emqx_router.erl

@@ -74,16 +74,14 @@
 %%--------------------------------------------------------------------
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?ROUTE_TAB, [
+    ok = mria:create_table(?ROUTE_TAB, [
                 {type, bag},
                 {rlog_shard, ?ROUTE_SHARD},
-                {ram_copies, [node()]},
+                {storage, ram_copies},
                 {record_name, route},
                 {attributes, record_info(fields, route)},
                 {storage_properties, [{ets, [{read_concurrency, true},
-                                             {write_concurrency, true}]}]}]);
+                                             {write_concurrency, true}]}]}]).
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?ROUTE_TAB, ram_copies).
 
 %%--------------------------------------------------------------------
 %% Start a router

apps/emqx/src/emqx_router_helper.erl

@@ -59,16 +59,13 @@
 %%--------------------------------------------------------------------
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?ROUTING_NODE, [
+    ok = mria:create_table(?ROUTING_NODE, [
                 {type, set},
                 {rlog_shard, ?ROUTE_SHARD},
-                {ram_copies, [node()]},
+                {storage, ram_copies},
                 {record_name, routing_node},
                 {attributes, record_info(fields, routing_node)},
-                {storage_properties, [{ets, [{read_concurrency, true}]}]}]);
+                {storage_properties, [{ets, [{read_concurrency, true}]}]}]).
-
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?ROUTING_NODE, ram_copies).
 
 %%--------------------------------------------------------------------
 %% API

apps/emqx/src/emqx_shared_sub.erl

@@ -85,15 +85,12 @@
 %%--------------------------------------------------------------------
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?TAB, [
+    ok = mria:create_table(?TAB, [
                 {type, bag},
                 {rlog_shard, ?SHARED_SUB_SHARD},
-                {ram_copies, [node()]},
+                {storage, ram_copies},
                 {record_name, emqx_shared_subscription},
-                {attributes, record_info(fields, emqx_shared_subscription)}]);
+                {attributes, record_info(fields, emqx_shared_subscription)}]).
-
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?TAB, ram_copies).
 
 %%--------------------------------------------------------------------
 %% API

apps/emqx/src/emqx_trie.erl

@@ -61,16 +61,13 @@ mnesia(boot) ->
     StoreProps = [{ets, [{read_concurrency, true},
                          {write_concurrency, true}
                         ]}],
-    ok = ekka_mnesia:create_table(?TRIE, [
+    ok = mria:create_table(?TRIE, [
                 {rlog_shard, ?ROUTE_SHARD},
-                {ram_copies, [node()]},
+                {storage, ram_copies},
                 {record_name, ?TRIE},
                 {attributes, record_info(fields, ?TRIE)},
                 {type, ordered_set},
-                {storage_properties, StoreProps}]);
+                {storage_properties, StoreProps}]).
-mnesia(copy) ->
-    %% Copy topics table
-    ok = ekka_mnesia:copy_table(?TRIE, ram_copies).
 
 %%--------------------------------------------------------------------
 %% Topics APIs

apps/emqx_authn/src/enhanced_authn/emqx_enhanced_authn_scram_mnesia.erl

@@ -63,15 +63,12 @@
 %% @doc Create or replicate tables.
 -spec(mnesia(boot | copy) -> ok).
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?TAB, [
+    ok = mria:create_table(?TAB, [
                 {rlog_shard, ?AUTH_SHARD},
-                {disc_copies, [node()]},
+                {storage, disc_copies},
                 {record_name, user_info},
                 {attributes, record_info(fields, user_info)},
-                {storage_properties, [{ets, [{read_concurrency, true}]}]}]);
+                {storage_properties, [{ets, [{read_concurrency, true}]}]}]).
-
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?TAB, disc_copies).
 
 %%------------------------------------------------------------------------------
 %% Hocon Schema

apps/emqx_authn/src/simple_authn/emqx_authn_mnesia.erl

@@ -70,15 +70,12 @@
 %% @doc Create or replicate tables.
 -spec(mnesia(boot | copy) -> ok).
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?TAB, [
+    ok = mria:create_table(?TAB, [
                 {rlog_shard, ?AUTH_SHARD},
-                {disc_copies, [node()]},
+                {storage, disc_copies},
                 {record_name, user_info},
                 {attributes, record_info(fields, user_info)},
-                {storage_properties, [{ets, [{read_concurrency, true}]}]}]);
+                {storage_properties, [{ets, [{read_concurrency, true}]}]}]).
-
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?TAB, disc_copies).
 
 %%------------------------------------------------------------------------------
 %% Hocon Schema

apps/emqx_authz/src/emqx_authz_mnesia.erl

@@ -36,14 +36,12 @@
 
 -spec(mnesia(boot | copy) -> ok).
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?ACL_TABLE, [
+    ok = mria:create_table(?ACL_TABLE, [
             {type, ordered_set},
             {rlog_shard, ?ACL_SHARDED},
-            {disc_copies, [node()]},
+            {storage, disc_copies},
             {attributes, record_info(fields, ?ACL_TABLE)},
-            {storage_properties, [{ets, [{read_concurrency, true}]}]}]);
+            {storage_properties, [{ets, [{read_concurrency, true}]}]}]).
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?ACL_TABLE, disc_copies).
 
 description() ->
     "AuthZ with Mnesia".

apps/emqx_dashboard/src/emqx_dashboard_admin.erl

@@ -50,16 +50,14 @@
 %%--------------------------------------------------------------------
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(mqtt_admin, [
+    ok = mria:create_table(mqtt_admin, [
                 {type, set},
                 {rlog_shard, ?DASHBOARD_SHARD},
-                {disc_copies, [node()]},
+                {storage, disc_copies},
                 {record_name, mqtt_admin},
                 {attributes, record_info(fields, mqtt_admin)},
                 {storage_properties, [{ets, [{read_concurrency, true},
-                                             {write_concurrency, true}]}]}]);
+                                             {write_concurrency, true}]}]}]).
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(mqtt_admin, disc_copies).
 
 %%--------------------------------------------------------------------
 %% API

apps/emqx_dashboard/src/emqx_dashboard_collection.erl

@@ -41,14 +41,12 @@
 -define(EXPIRE_INTERVAL, 86400000 * 7).
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(emqx_collect, [
+    ok = mria:create_table(emqx_collect, [
         {type, set},
         {local_content, true},
-        {disc_only_copies, [node()]},
+        {storage, disc_only_copies},
         {record_name, mqtt_collect},
-        {attributes, record_info(fields, mqtt_collect)}]);
+        {attributes, record_info(fields, mqtt_collect)}]).
-mnesia(copy) ->
-    mnesia:add_table_copy(emqx_collect, node(), disc_only_copies).
 
 start_link() ->
     gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

apps/emqx_dashboard/src/emqx_dashboard_token.erl

@@ -77,16 +77,14 @@ destroy_by_username(Username) ->
     do_destroy_by_username(Username).
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?TAB, [
+    ok = mria:create_table(?TAB, [
                 {type, set},
                 {rlog_shard, ?DASHBOARD_SHARD},
-                {disc_copies, [node()]},
+                {storage, disc_copies},
                 {record_name, mqtt_admin_jwt},
                 {attributes, record_info(fields, mqtt_admin_jwt)},
                 {storage_properties, [{ets, [{read_concurrency, true},
-                                             {write_concurrency, true}]}]}]);
+                                             {write_concurrency, true}]}]}]).
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?TAB, disc_copies).
 
 %%--------------------------------------------------------------------
 %% jwt apply

apps/emqx_gateway/src/emqx_gateway_cm_registry.erl

@@ -89,16 +89,15 @@ record(ClientId, ChanPid) ->
 
 init([Type]) ->
     Tab = tabname(Type),
-    ok = ekka_mnesia:create_table(Tab, [
+    ok = mria:create_table(Tab, [
                 {type, bag},
                 {rlog_shard, ?CM_SHARD},
-                {ram_copies, [node()]},
+                {storage, ram_copies},
                 {record_name, channel},
                 {attributes, record_info(fields, channel)},
                 {storage_properties, [{ets, [{read_concurrency, true},
                                              {write_concurrency, true}]}]}]),
-    ok = ekka_mnesia:copy_table(Tab, ram_copies),
+    ok = mria:wait_for_tables([Tab]),
-    %%ok = ekka_rlog:wait_for_shards([?CM_SHARD], infinity),
     ok = ekka:monitor(membership),
     {ok, #{type => Type}}.
 

apps/emqx_gateway/src/mqttsn/emqx_sn_registry.erl

@@ -63,13 +63,10 @@
 %mnesia(boot) ->
 %    %% Optimize storage
 %    StoreProps = [{ets, [{read_concurrency, true}]}],
-%    ok = ekka_mnesia:create_table(?MODULE, [
+%    ok = mria:create_table(?MODULE, [
 %            {attributes, record_info(fields, emqx_sn_registry)},
 %            {ram_copies, [node()]},
-%            {storage_properties, StoreProps}]);
+%            {storage_properties, StoreProps}]).
-%
-%mnesia(copy) ->
-%    ok = ekka_mnesia:copy_table(?MODULE, ram_copies).
 
 -type registry() :: {Tab :: atom(),
                      RegistryPid :: pid()}.
@@ -141,15 +138,14 @@ init([InstaId, PredefTopics]) ->
     %% {ClientId, TopicId}   -> TopicName
     %% {ClientId, TopicName} -> TopicId
     Tab = name(InstaId),
-    ok = ekka_mnesia:create_table(Tab, [
+    ok = mria:create_table(Tab, [
-                {ram_copies, [node()]},
+                {storage, ram_copies},
                 {record_name, emqx_sn_registry},
                 {attributes, record_info(fields, emqx_sn_registry)},
                 {storage_properties, [{ets, [{read_concurrency, true}]}]},
                 {rlog_shard, ?SN_SHARD}
                ]),
-    ok = ekka_mnesia:copy_table(Tab, ram_copies),
+    ok = mria:wait_for_tables([Tab]),
-    ok = ekka_rlog:wait_for_shards([?SN_SHARD], infinity),
     % FIXME:
     %ok = ekka_rlog:wait_for_shards([?CM_SHARD], infinity),
     MaxPredefId = lists:foldl(

apps/emqx_machine/src/emqx_cluster_rpc.erl

@@ -45,21 +45,18 @@
 %%% API
 %%%===================================================================
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?CLUSTER_MFA, [
+    ok = mria:create_table(?CLUSTER_MFA, [
         {type, ordered_set},
         {rlog_shard, ?EMQX_MACHINE_SHARD},
-        {disc_copies, [node()]},
+        {storage, disc_copies},
         {record_name, cluster_rpc_mfa},
         {attributes, record_info(fields, cluster_rpc_mfa)}]),
-    ok = ekka_mnesia:create_table(?CLUSTER_COMMIT, [
+    ok = mria:create_table(?CLUSTER_COMMIT, [
         {type, set},
         {rlog_shard, ?EMQX_MACHINE_SHARD},
-        {disc_copies, [node()]},
+        {storage, disc_copies},
         {record_name, cluster_rpc_commit},
-        {attributes, record_info(fields, cluster_rpc_commit)}]);
+        {attributes, record_info(fields, cluster_rpc_commit)}]).
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(cluster_rpc_mfa, disc_copies),
-    ok = ekka_mnesia:copy_table(cluster_rpc_commit, disc_copies).
 
 start_link() ->
     start_link(node(), ?MODULE, get_retry_ms()).

apps/emqx_modules/src/emqx_delayed.erl

@@ -64,14 +64,12 @@
 %% Mnesia bootstrap
 %%--------------------------------------------------------------------
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?TAB, [
+    ok = mria:create_table(?TAB, [
                 {type, ordered_set},
-                {disc_copies, [node()]},
+                {storage, disc_copies},
                 {local_content, true},
                 {record_name, delayed_message},
-                {attributes, record_info(fields, delayed_message)}]);
+                {attributes, record_info(fields, delayed_message)}]).
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?TAB, disc_copies).
 
 %%--------------------------------------------------------------------
 %% Hooks

apps/emqx_modules/src/emqx_telemetry.erl

@@ -91,14 +91,12 @@
 %%--------------------------------------------------------------------
 
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?TELEMETRY,
+    ok = mria:create_table(?TELEMETRY,
              [{type, set},
-              {disc_copies, [node()]},
+              {storage, disc_copies},
               {local_content, true},
               {record_name, telemetry},
-              {attributes, record_info(fields, telemetry)}]);
+              {attributes, record_info(fields, telemetry)}]).
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?TELEMETRY, disc_copies).
 
 %%--------------------------------------------------------------------
 %% API

apps/emqx_modules/test/emqx_delayed_SUITE.erl

@@ -35,7 +35,7 @@ all() ->
     emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
-    ekka_mnesia:start(),
+    mria:start(),
     ok = emqx_delayed:mnesia(boot),
     emqx_common_test_helpers:start_apps([emqx_modules]),
     Config.

apps/emqx_modules/test/emqx_telemetry_SUITE.erl

@@ -28,7 +28,7 @@
 all() -> emqx_common_test_helpers:all(?MODULE).
 
 init_per_suite(Config) ->
-    ok = ekka_mnesia:start(),
+    ok = mria:start(),
     ok = emqx_telemetry:mnesia(boot),
     emqx_common_test_helpers:start_apps([emqx_modules]),
     Config.

apps/emqx_prometheus/src/emqx_prometheus.erl

@@ -590,6 +590,6 @@ emqx_cluster() ->
     ].
 
 emqx_cluster_data() ->
-    #{running_nodes := Running, stopped_nodes := Stopped} = ekka_mnesia:cluster_info(),
+    #{running_nodes := Running, stopped_nodes := Stopped} = mria:cluster_info(),
     [{nodes_running, length(Running)},
      {nodes_stopped, length(Stopped)}].

apps/emqx_psk/src/emqx_psk.erl

@@ -64,16 +64,13 @@
 %% @doc Create or replicate tables.
 -spec(mnesia(boot | copy) -> ok).
 mnesia(boot) ->
-    ok = ekka_mnesia:create_table(?TAB, [
+    ok = mria:create_table(?TAB, [
                 {rlog_shard, ?PSK_SHARD},
                 {type, ordered_set},
-                {disc_copies, [node()]},
+                {storage, disc_copies},
                 {record_name, psk_entry},
                 {attributes, record_info(fields, psk_entry)},
-                {storage_properties, [{ets, [{read_concurrency, true}]}]}]);
+                {storage_properties, [{ets, [{read_concurrency, true}]}]}]).
-
-mnesia(copy) ->
-    ok = ekka_mnesia:copy_table(?TAB, disc_copies).
 
 %%------------------------------------------------------------------------------
 %% APIs

apps/emqx_retainer/src/emqx_retainer_mnesia.erl

@@ -52,14 +52,13 @@ create_resource(#{storage_type := StorageType}) ->
                          {read_concurrency, true},
                          {write_concurrency, true}]},
                   {dets, [{auto_save, 1000}]}],
-    ok = ekka_mnesia:create_table(?TAB, [
+    ok = mria:create_table(?TAB, [
                 {type, set},
                 {rlog_shard, ?RETAINER_SHARD},
-                {Copies, [node()]},
+                {storage, Copies},
                 {record_name, retained},
                 {attributes, record_info(fields, retained)},
                 {storage_properties, StoreProps}]),
-    ok = ekka_mnesia:copy_table(?TAB, Copies),
     ok = ekka_rlog:wait_for_shards([?RETAINER_SHARD], infinity),
     case mnesia:table_info(?TAB, storage_type) of
         Copies -> ok;