Merge pull request #292 from cloudamqp/4_0

Ensure Mnesia is running when Khepri is used as main metadata store

README.md

@@ -33,7 +33,12 @@ of some kind.
 
 The most recent release of this plugin targets RabbitMQ 4.0.x.
 
-This plugin currently only supports Mnesia for metadata store (do not use it with Khepri).
+This plugin can be enabled on a RabbitMQ cluster that uses either Mnesia or Khepri as metadata store.
+
+Warning: the plugin must be disabled during Khepri migration. One
+needs to disable this plugin before enabling the `khepri_db` feature
+flag and enable it after. This will result in losing all previous
+delayed messages.
 
 ## Supported Erlang/OTP Versions
 

src/rabbit_delayed_message.erl

@@ -85,6 +85,13 @@ delay_message(Exchange, Message, Delay) ->
                     infinity).
 
 setup_mnesia() ->
+    case rabbit_khepri:is_enabled() of
+        true ->
+            ensure_mnesia_running();
+        false ->
+            %% Mnesia should already be running
+            ok
+    end,
     _ = mnesia:create_table(?TABLE_NAME, [{record_name, delay_entry},
                                           {attributes,
                                            record_info(fields, delay_entry)},
@@ -97,6 +104,25 @@ setup_mnesia() ->
                                                 {disc_copies, [node()]}]),
     rabbit_table:wait([?TABLE_NAME, ?INDEX_TABLE_NAME]).
 
+ensure_mnesia_running() ->
+    case rabbit_mnesia:is_running() of
+        false ->
+            ensure_mnesia_disc_schema(),
+            rabbit_mnesia:start_mnesia(_CheckConsistency = false);
+        true ->
+            ok
+    end.
+
+ensure_mnesia_disc_schema() ->
+    case mnesia:system_info(use_dir) of
+        true ->
+            %% There is a disc schema already
+            ok;
+        false ->
+            rabbit_misc:ensure_ok(mnesia:create_schema([node()]),
+                                  {?MODULE, cannot_create_mnesia_schema})
+    end.
+
 disable_plugin() ->
     _ = mnesia:delete_table(?INDEX_TABLE_NAME),
     _ = mnesia:delete_table(?TABLE_NAME),