%%%------------------------------------------------------------------- %% @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).