Merge pull request #156 from rabbitmq/configurable-writer-module

Make writer and replica impl modules pluggable

Author: kjnilsson

src/osiris.erl

@@ -23,8 +23,12 @@
          update_retention/2,
          start_cluster/1,
          stop_cluster/1,
+
          start_writer/1,
          start_replica/2,
+         stop_member/2,
+         delete_member/2,
+
          delete_cluster/1,
          configure_logger/1,
          get_stats/1]).
@@ -103,7 +107,7 @@ start_cluster(Config00 = #{name := Name}) ->
     true = osiris_util:validate_base64uri(Name),
     %% ensure reference is set
     Config0 = maps:merge(#{reference => Name}, Config00),
-    case osiris_writer:start(Config0) of
+    case start_writer(Config0) of
         {ok, Pid} ->
             Config = Config0#{leader_pid => Pid},
             case start_replicas(Config) of
@@ -115,22 +119,41 @@ start_cluster(Config00 = #{name := Name}) ->
     end.
 
 stop_cluster(Config) ->
-    ok = osiris_writer:stop(Config),
-    [ok = osiris_replica:stop(N, Config)
+    WriterNode = maps:get(leader_node, Config),
+    ok = stop_member(WriterNode, Config),
+    [ok = stop_member(N, Config)
      || N <- maps:get(replica_nodes, Config)],
     ok.
 
 -spec delete_cluster(config()) -> ok.
 delete_cluster(Config) ->
-    [ok = osiris_replica:delete(R, Config)
-     || R <- maps:get(replica_nodes, Config)],
-    ok = osiris_writer:delete(Config).
+    [ok = delete_member(N, Config)
+     || N <- maps:get(replica_nodes, Config)],
+    WriterNode = maps:get(leader_node, Config),
+    ok = delete_member(WriterNode, Config).
 
+-spec start_writer(osiris:config()) ->
+    supervisor:startchild_ret().
 start_writer(Config) ->
-    osiris_writer:start(Config).
+    Mod = get_writer_module(Config),
+    Node = maps:get(leader_node, Config),
+    osiris_member:start(Mod, Node, Config).
 
-start_replica(Replica, Config) ->
-    osiris_replica:start(Replica, Config).
+-spec start_replica(node(), osiris:config()) ->
+    supervisor:startchild_ret().
+start_replica(Node, Config) ->
+    Mod = maps:get(replica_mod, Config, osiris_replica),
+    osiris_member:start(Mod, Node, Config).
+
+-spec stop_member(node(), osiris:config()) ->
+    ok | {error, not_found}.
+stop_member(Node, Config) ->
+    osiris_member:stop(Node, Config).
+
+-spec delete_member(node(), osiris:config()) ->
+    ok | {error, not_found}.
+delete_member(Node, Config) ->
+    osiris_member:delete(Node, Config).
 
 -spec write(Pid :: pid(), Data :: data()) -> ok.
 write(Pid, Data) ->
@@ -258,7 +281,7 @@ start_replicas(_Config, [], ReplicaPids) ->
     {ok, ReplicaPids};
 start_replicas(Config, [Node | Nodes], ReplicaPids) ->
     try
-        case osiris_replica:start(Node, Config) of
+        case start_replica(Node, Config) of
             {ok, Pid} ->
                 start_replicas(Config, Nodes, [Pid | ReplicaPids]);
             {ok, Pid, _} ->
@@ -293,3 +316,6 @@ get_stats(Pid)
       last_chunk_id => osiris_log_shared:last_chunk_id(Shared)};
 get_stats(Pid) when is_pid(Pid) ->
     erpc:call(node(Pid), ?MODULE, ?FUNCTION_NAME, [Pid]).
+
+get_writer_module(Config) ->
+    maps:get(writer_mod, Config, osiris_writer).

src/osiris_member.erl

@@ -0,0 +1,29 @@
+-module(osiris_member).
+
+
+-include("osiris.hrl").
+-export([start/3,
+         stop/2,
+         delete/2]).
+
+%% Callbacks
+
+-callback start(node(), osiris:config()) ->
+    supervisor:startchild_ret().
+
+%% API
+
+-spec start(module(), node(), osiris:config()) ->
+    supervisor:startchild_ret().
+start(Mod, Node, Config) ->
+    Mod:start(Node, Config).
+
+-spec stop(node(), osiris:config()) ->
+    ok | {error, not_found}.
+stop(Node, Config) ->
+    ?SUP:stop_child(Node, Config).
+
+-spec delete(node(), osiris:config()) ->
+    ok | {error, term()}.
+delete(Node, Config) ->
+    ?SUP:delete_child(Node, Config).

src/osiris_replica.erl

@@ -8,6 +8,7 @@
 -module(osiris_replica).
 
 -behaviour(gen_server).
+-behaviour(osiris_member).
 
 -include("osiris.hrl").
 
@@ -17,9 +18,8 @@
 
 %% API functions
 -export([start/2,
-         start_link/1,
-         stop/2,
-         delete/2]).
+         start_link/1
+        ]).
 %% Test
 -export([get_port/1, combine_ips_hosts/4]).
 %% gen_server callbacks
@@ -86,6 +86,8 @@
 %%% API functions
 %%%===================================================================
 
+-spec start(node(), Config :: osiris:config()) ->
+    supervisor:startchild_ret().
 start(Node, Config = #{name := Name}) when ?IS_STRING(Name) ->
     case supervisor:start_child({?SUP, Node},
                                 #{id => Name,
@@ -105,12 +107,6 @@ start(Node, Config = #{name := Name}) when ?IS_STRING(Name) ->
             Err
     end.
 
-stop(Node, #{name := Name}) ->
-    ?SUP:stop_child(Node, Name).
-
-delete(Node, Config = #{}) ->
-    ?SUP:delete_child(Node, Config).
-
 %%--------------------------------------------------------------------
 %% @doc
 %% Starts the server

src/osiris_server_sup.erl

@@ -9,6 +9,7 @@
 
 -behaviour(supervisor).
 
+-include("osiris.hrl").
 -export([start_link/0]).
 -export([init/1,
          stop_child/2,
@@ -21,7 +22,9 @@ init([]) ->
     Procs = [],
     {ok, {{one_for_one, 1, 5}, Procs}}.
 
-stop_child(Node, Name) ->
+stop_child(Node, #{name := Name}) ->
+    stop_child(Node, Name);
+stop_child(Node, Name) when ?IS_STRING(Name) ->
     try
         %% as replicas are temporary we don't have to explicitly
         %% delete them

src/osiris_writer.erl

@@ -9,6 +9,7 @@
 -module(osiris_writer).
 
 -behaviour(gen_batch_server).
+-behaviour(osiris_member).
 
 -include("osiris.hrl").
 
@@ -33,6 +34,11 @@
          stop/1,
          delete/1]).
 
+%% osiris_member impl
+-export([start/2,
+         stop/2,
+         delete/2]).
+
 -define(C_COMMITTED_OFFSET, ?C_NUM_LOG_FIELDS + 1).
 -define(C_READERS, ?C_NUM_LOG_FIELDS + 2).
 -define(C_EPOCH, ?C_NUM_LOG_FIELDS + 3).
@@ -73,18 +79,33 @@
 
 -export_type([state/0]).
 
-start(Config = #{name := Name, leader_node := Leader}) ->
-    supervisor:start_child({?SUP, Leader},
+-spec start(node(), Config :: osiris:config()) ->
+    supervisor:startchild_ret().
+start(Node, #{name := Name, leader_node := Node} = Config) ->
+    supervisor:start_child({?SUP, Node},
                            #{id => osiris_util:normalise_name(Name),
                              start => {?MODULE, start_link, [Config]},
                              restart => temporary,
                              shutdown => 5000,
                              type => worker}).
 
-stop(#{name := Name, leader_node := Node}) ->
-    %% stop child handles name normalisation
-    ?SUP:stop_child(Node, Name).
+-spec stop(node(), osiris:config()) ->
+    ok | {error, not_found}.
+stop(Node, #{leader_node := Node} = Config) ->
+    ?SUP:stop_child(Node, Config).
+
+-spec delete(node(), osiris:config()) ->
+    ok | {error, term()}.
+delete(Node, #{leader_node := Node} = Config) ->
+    ?SUP:delete_child(Node, Config).
 
+%% backwards compat
+start(#{leader_node := LeaderNode} = Config) ->
+    start(LeaderNode, Config).
+%% backwards compat
+stop(#{leader_node := Node} = Config) ->
+    stop(Node, Config).
+%% backwards compat
 delete(#{leader_node := Node} = Config) ->
     ?SUP:delete_child(Node, Config).