eredis_streams

%%%-------------------------------------------------------------------
%% @doc eota_streams top level supervisor.
%% @end
%%%-------------------------------------------------------------------

-module(eredis_streams_stream_sup).

-behaviour(supervisor).

-export([start_link/1]).
-export([init/1]).
-export([process_name/1]).

start_link(#{name := StreamName} = StreamConfig) ->
  supervisor:start_link({local, process_name(StreamName)}, ?MODULE, StreamConfig).

init(StreamConfig) ->
  SupFlags = #{strategy => one_for_one, intensity => 10, period => 1},
  WorkersConfig = maps:get(workers, StreamConfig, #{}),
  WorkersCnt = maps:get(workers_cnt, WorkersConfig, 1),
  ChildSpecs =
    [
      [worker_spec(StreamConfig, Num) || Num <- lists:seq(1, WorkersCnt)],
      failover_spec(StreamConfig)
    ],
  {ok, {SupFlags, lists:flatten(ChildSpecs)}}.

%% internal functions

worker_spec(#{name := StreamName} = StreamConfig, Num) ->
  #{
    id => {eredis_streams_worker, StreamName, Num},
    start => {eredis_streams_worker, start_link, [StreamConfig#{worker_num => Num}]},
    restart => permanent,
    shutdown => shutdown_timeout(StreamConfig),
    type => worker,
    modules => [eredis_streams_worker]
  }.

failover_spec(#{name := StreamName} = StreamConfig) ->
  #{
    id => {eredis_streams_failover, StreamName},
    start => {eredis_streams_failover, start_link, [StreamConfig]},
    restart => permanent,
    shutdown => shutdown_timeout(StreamConfig),
    type => worker,
    modules => [eredis_streams_failover]
  }.

process_name(StreamName) ->
  binary_to_atom(iolist_to_binary([["eredis_streams_", StreamName, "_sup"]]), utf8).

shutdown_timeout(Params) -> maps:get(shutdown, Params, 5000).