defmodule ConsumerSupervisor do
@moduledoc ~S"""
A supervisor that starts children as events flow in.
A `ConsumerSupervisor` can be used as the consumer in a `GenStage` pipeline.
A new child process will be started per event, where the event is appended
to the arguments in the child specification.
A `ConsumerSupervisor` can be attached to a producer by returning
`:subscribe_to` from `c:init/1` or explicitly with `GenStage.sync_subscribe/3`
and `GenStage.async_subscribe/2`.
Once subscribed, the supervisor will ask the producer for `:max_demand` events
and start child processes as events arrive. As child processes terminate, the
supervisor will accumulate demand and request more events once `:min_demand`
is reached. This allows the `ConsumerSupervisor` to work similar to a pool,
except a child process is started per event. The minimum amount of concurrent
children per producer is specified by `:min_demand` and the maximum is given
by `:max_demand`.
## Example
Let's define a GenStage consumer as a `ConsumerSupervisor` that subscribes
to a producer named `Producer` and starts a new process for each event
received from the producer. Each new process will be started by calling
`Printer.start_link/1`, which simply starts a task that will print the
incoming event to the terminal.
defmodule Consumer do
use ConsumerSupervisor
def start_link(arg) do
ConsumerSupervisor.start_link(__MODULE__, arg)
end
def init(_arg) do
# Note: By default the restart for a child is set to :permanent
# which is not supported in ConsumerSupervisor. You need to explicitly
# set the :restart option either to :temporary or :transient.
children = [%{id: Printer, start: {Printer, :start_link, []}, restart: :transient}]
opts = [strategy: :one_for_one, subscribe_to: [{Producer, max_demand: 50}]]
ConsumerSupervisor.init(children, opts)
end
end
Then in the `Printer` module:
defmodule Printer do
def start_link(event) do
# Note: this function must return the format of `{:ok, pid}` and like
# all children started by a Supervisor, the process must be linked
# back to the supervisor (if you use `Task.start_link/1` then both
# these requirements are met automatically)
Task.start_link(fn ->
IO.inspect({self(), event})
end)
end
end
Similar to `Supervisor`, `ConsumerSupervisor` also provides `start_link/3`,
which allows developers to start a supervisor with the help of a callback
module.
## Name Registration
A supervisor is bound to the same name registration rules as a `GenServer`.
Read more about it in the `GenServer` docs.
"""
@behaviour GenStage
@typedoc "Options used by the `start*` functions"
@type option ::
{:registry, atom}
| {:name, Supervisor.name()}
| {:strategy, Supervisor.Spec.strategy()}
| {:max_restarts, non_neg_integer}
| {:max_seconds, non_neg_integer}
| {:subscribe_to, [GenStage.stage() | {GenStage.stage(), keyword()}]}
@doc """
Callback invoked to start the supervisor and during hot code upgrades.
## Options
* `:strategy` - the restart strategy option. Only `:one_for_one`
is supported by consumer supervisors.
* `:max_restarts` - the maximum amount of restarts allowed in
a time frame. Defaults to 3 times.
* `:max_seconds` - the time frame in which `:max_restarts` applies
in seconds. Defaults to 5 seconds.
* `:subscribe_to` - a list of producers to subscribe to. Each element
represents the producer or a tuple with the producer and the subscription
options, for example, `[Producer]` or `[{Producer, max_demand: 20, min_demand: 10}]`.
"""
@callback init(args :: term) ::
{:ok, [:supervisor.child_spec()], options :: keyword()}
| :ignore
defstruct [
:name,
:mod,
:args,
:template,
:max_restarts,
:max_seconds,
:strategy,
children: %{},
producers: %{},
restarts: [],
restarting: 0
]
@doc false
defmacro __using__(opts) do
quote location: :keep, bind_quoted: [opts: opts] do
@behaviour ConsumerSupervisor
import Supervisor.Spec
@doc false
def child_spec(arg) do
default = %{
id: __MODULE__,
start: {__MODULE__, :start_link, [arg]},
type: :supervisor
}
Supervisor.child_spec(default, unquote(Macro.escape(opts)))
end
defoverridable child_spec: 1
@doc false
def init(arg)
end
end
defmodule Default do
@moduledoc false
def init(args) do
args
end
end
@doc """
Starts a supervisor with the given children.
A strategy is required to be given as an option. Furthermore,
the `:max_restarts`, `:max_seconds`, and `:subscribe_to`
values can be configured as described in the documentation for the
`c:init/1` callback.
The options can also be used to register a supervisor name.
The supported values are described under the "Name Registration"
section in the `GenServer` module docs.
The child processes specified in `children` will be started by appending
the event to process to the existing function arguments in the child specification.
Note that the consumer supervisor is linked to the parent process
and will exit not only on crashes but also if the parent process
exits with `:normal` reason.
"""
@spec start_link([Supervisor.Spec.spec() | Supervisor.child_spec()], [option]) ::
Supervisor.on_start()
def start_link(children, options) when is_list(children) do
{sup_options, start_options} =
Keyword.split(options, [:strategy, :max_restarts, :max_seconds, :subscribe_to])
start_link(Default, init(children, sup_options), start_options)
end
@doc """
Starts a consumer supervisor module with the given `args`.
To start the supervisor, the `c:init/1` callback will be invoked in the given
module, with `args` passed to it. The `c:init/1` callback must return a
supervision specification which can be created with the help of the
`Supervisor` module.
If the `c:init/1` callback returns `:ignore`, this function returns
`:ignore` as well and the supervisor terminates with reason `:normal`.
If it fails or returns an incorrect value, this function returns
`{:error, term}` where `term` is a term with information about the
error, and the supervisor terminates with reason `term`.
The `:name` option can also be given in order to register a supervisor
name. The supported values are described under the "Name Registration"
section in the `GenServer` module docs.
"""
@spec start_link(module, any) :: Supervisor.on_start()
@spec start_link(module, any, [option]) :: Supervisor.on_start()
def start_link(mod, args, opts \\ []) do
GenStage.start_link(__MODULE__, {mod, args, opts[:name]}, opts)
end
@doc """
Starts a child in the consumer supervisor.
The child process will be started by appending the given list of
`args` to the existing function arguments in the child specification.
This child is started separately from any producer and does not
count towards the demand of any of them.
If the child process starts, function returns `{:ok, child}` or
`{:ok, child, info}`, the pid is added to the supervisor, and the
function returns the same value.
If the child process start function returns `:ignore`, an error tuple,
or an erroneous value, or if it fails, the child is discarded and
`:ignore` or `{:error, error}` where `error` is a term containing
information about the error is returned.
"""
@spec start_child(Supervisor.supervisor(), [term]) :: Supervisor.on_start_child()
def start_child(supervisor, args) when is_list(args) do
call(supervisor, {:start_child, args})
end
@doc """
Terminates the given child pid.
If successful, the function returns `:ok`. If there is no
such pid, the function returns `{:error, :not_found}`.
"""
@spec terminate_child(Supervisor.supervisor(), pid) :: :ok | {:error, :not_found}
def terminate_child(supervisor, pid) when is_pid(pid) do
call(supervisor, {:terminate_child, pid})
end
@doc """
Returns a list with information about all children.
Note that calling this function when supervising a large number
of children under low memory conditions can cause an out of memory
exception.
This function returns a list of tuples containing:
* `id` - as defined in the child specification but is always
set to `:undefined` for consumer supervisors
* `child` - the pid of the corresponding child process or the
atom `:restarting` if the process is about to be restarted
* `type` - `:worker` or `:supervisor` as defined in the child
specification
* `modules` - as defined in the child specification
"""
@spec which_children(Supervisor.supervisor()) :: [
{:undefined, pid | :restarting, :worker | :supervisor, :dynamic | [module()]}
]
def which_children(supervisor) do
call(supervisor, :which_children)
end
@doc """
Returns a map containing count values for the supervisor.
The map contains the following keys:
* `:specs` - always `1` as consumer supervisors have a single specification
* `:active` - the count of all actively running child processes managed by
this supervisor
* `:supervisors` - the count of all supervisors whether or not the child
process is still alive
* `:workers` - the count of all workers, whether or not the child process
is still alive
"""
@spec count_children(Supervisor.supervisor()) :: %{
specs: non_neg_integer,
active: non_neg_integer,
supervisors: non_neg_integer,
workers: non_neg_integer
}
def count_children(supervisor) do
call(supervisor, :count_children)
end
@doc """
Receives a template to initialize and a set of options.
This is typically invoked at the end of the `c:init/1` callback of module-based supervisors.
This function returns a the child specification and the supervisor flags.
## Examples
Using the child specification changes introduced in Elixir 1.5:
defmodule MyConsumerSupervisor do
use ConsumerSupervisor
def start_link(arg) do
ConsumerSupervisor.start_link(__MODULE__, arg)
end
def init(_arg) do
ConsumerSupervisor.init([MyConsumer], strategy: :one_for_one, subscribe_to: MyProducer)
end
end
"""
def init([{_, _, _, _, _, _} = template], opts) do
{:ok, [template], opts}
end
def init([template], opts) when is_tuple(template) or is_map(template) or is_atom(template) do
{:ok, {_, [template]}} = Supervisor.init([template], opts)
{:ok, [template], opts}
end
@compile {:inline, call: 2}
defp call(supervisor, req) do
GenStage.call(supervisor, req, :infinity)
end
## Callbacks
@impl true
def init({mod, args, name}) do
Process.put(:"$initial_call", {:supervisor, mod, 1})
Process.flag(:trap_exit, true)
case mod.init(args) do
{:ok, children, opts} ->
case validate_specs(children) do
:ok ->
state = %ConsumerSupervisor{mod: mod, args: args, name: name || {self(), mod}}
case init(state, children, opts) do
{:ok, state, opts} -> {:consumer, state, opts}
{:error, message} -> {:stop, {:bad_opts, message}}
end
{:error, message} ->
{:stop, {:bad_specs, message}}
end
:ignore ->
:ignore
other ->
{:stop, {:bad_return_value, other}}
end
end
defp init(state, [child], opts) when is_list(opts) do
{strategy, opts} = Keyword.pop(opts, :strategy)
{max_restarts, opts} = Keyword.pop(opts, :max_restarts, 3)
{max_seconds, opts} = Keyword.pop(opts, :max_seconds, 5)
template = normalize_template(child)
with :ok <- validate_strategy(strategy),
:ok <- validate_restarts(max_restarts),
:ok <- validate_seconds(max_seconds),
:ok <- validate_template(template) do
state = %{
state
| template: template,
strategy: strategy,
max_restarts: max_restarts,
max_seconds: max_seconds
}
{:ok, state, opts}
end
end
defp init(_state, [_], _opts) do
{:error, "supervisor's init expects a keywords list as options"}
end
defp validate_specs([_] = children) do
:supervisor.check_childspecs(children)
end
defp validate_specs(_children) do
{:error, "consumer supervisor expects a list with a single item as a template"}
end
defp validate_strategy(strategy) when strategy in [:one_for_one], do: :ok
defp validate_strategy(nil), do: {:error, "supervisor expects a strategy to be given"}
defp validate_strategy(_), do: {:error, "unknown supervision strategy for consumer supervisor"}
defp validate_restarts(restart) when is_integer(restart), do: :ok
defp validate_restarts(_), do: {:error, "max_restarts must be an integer"}
defp validate_seconds(seconds) when is_integer(seconds), do: :ok
defp validate_seconds(_), do: {:error, "max_seconds must be an integer"}
@impl true
def handle_subscribe(:producer, opts, {_, ref} = from, state) do
# GenStage checks these options before allowing susbcription
max = Keyword.get(opts, :max_demand, 1000)
min = Keyword.get(opts, :min_demand, div(max, 2))
GenStage.ask(from, max)
{:manual, put_in(state.producers[ref], {from, 0, 0, min, max})}
end
@impl true
def handle_cancel(_, {_, ref}, state) do
{:noreply, [], update_in(state.producers, &Map.delete(&1, ref))}
end
@impl true
def handle_events(events, {pid, ref} = from, state) do
%{template: child, children: children} = state
{new, errors} = start_events(events, from, child, 0, [], state)
new_children = Enum.into(new, children)
started = map_size(new_children) - map_size(children)
{:noreply, [], maybe_ask(ref, pid, started + errors, errors, new_children, state)}
end
defp start_events([extra | extras], from, child, errors, acc, state) do
{_, ref} = from
{_, {m, f, args}, restart, _, _, _} = child
args = args ++ [extra]
case start_child(m, f, args) do
{:ok, pid, _} when restart == :temporary ->
acc = [{pid, [ref | :undefined]} | acc]
start_events(extras, from, child, errors, acc, state)
{:ok, pid, _} ->
acc = [{pid, [ref | args]} | acc]
start_events(extras, from, child, errors, acc, state)
{:ok, pid} when restart == :temporary ->
acc = [{pid, [ref | :undefined]} | acc]
start_events(extras, from, child, errors, acc, state)
{:ok, pid} ->
acc = [{pid, [ref | args]} | acc]
start_events(extras, from, child, errors, acc, state)
:ignore ->
start_events(extras, from, child, errors + 1, acc, state)
{:error, reason} ->
:error_logger.error_msg(
~c"ConsumerSupervisor failed to start child from: ~tp with reason: ~tp~n",
[from, reason]
)
report_error(:start_error, reason, :undefined, args, child, state)
start_events(extras, from, child, errors + 1, acc, state)
end
end
defp start_events([], _, _, errors, acc, _) do
{acc, errors}
end
defp maybe_ask(ref, pid, events, down, children, state) do
%{producers: producers} = state
case producers do
%{^ref => {to, count, pending, min, max}} ->
if count + events > max do
:error_logger.error_msg(
~c"ConsumerSupervisor has received ~tp events in excess from: ~tp~n",
[count + events - max, {pid, ref}]
)
end
pending =
case pending + down do
ask when ask >= min ->
GenStage.ask(to, ask)
0
ask ->
ask
end
count = count + events - down
producers = Map.put(producers, ref, {to, count, pending, min, max})
%{state | children: children, producers: producers}
%{} ->
%{state | children: children}
end
end
@impl true
def handle_call(:which_children, _from, state) do
%{children: children, template: child} = state
{_, _, _, _, type, mods} = child
reply =
for {pid, args} <- children do
maybe_pid =
case args do
{:restarting, _} -> :restarting
_ -> pid
end
{:undefined, maybe_pid, type, mods}
end
{:reply, reply, [], state}
end
def handle_call(:count_children, _from, state) do
%{children: children, template: child, restarting: restarting} = state
{_, _, _, _, type, _} = child
specs = map_size(children)
active = specs - restarting
reply =
case type do
:supervisor ->
%{specs: 1, active: active, workers: 0, supervisors: specs}
:worker ->
%{specs: 1, active: active, workers: specs, supervisors: 0}
end
{:reply, reply, [], state}
end
def handle_call({:terminate_child, pid}, _from, %{children: children} = state) do
case children do
%{^pid => [producer | _] = info} ->
:ok = terminate_children(%{pid => info}, state)
{:reply, :ok, [], delete_child_and_maybe_ask(producer, pid, state)}
%{^pid => {:restarting, [producer | _]} = info} ->
:ok = terminate_children(%{pid => info}, state)
{:reply, :ok, [], delete_child_and_maybe_ask(producer, pid, state)}
%{} ->
{:reply, {:error, :not_found}, [], state}
end
end
def handle_call({:start_child, extra}, _from, %{template: child} = state) do
handle_start_child(child, extra, state)
end
defp handle_start_child({_, {m, f, args}, restart, _, _, _}, extra, state) do
args = args ++ extra
case reply = start_child(m, f, args) do
{:ok, pid, _} ->
{:reply, reply, [], save_child(restart, :dynamic, pid, args, state)}
{:ok, pid} ->
{:reply, reply, [], save_child(restart, :dynamic, pid, args, state)}
_ ->
{:reply, reply, [], state}
end
end
defp start_child(m, f, a) do
try do
apply(m, f, a)
catch
kind, reason ->
{:error, exit_reason(kind, reason, __STACKTRACE__)}
else
{:ok, pid, extra} when is_pid(pid) -> {:ok, pid, extra}
{:ok, pid} when is_pid(pid) -> {:ok, pid}
:ignore -> :ignore
{:error, _} = error -> error
other -> {:error, other}
end
end
defp save_child(:temporary, producer, pid, _, state),
do: put_in(state.children[pid], [producer | :undefined])
defp save_child(_, producer, pid, args, state),
do: put_in(state.children[pid], [producer | args])
defp exit_reason(:exit, reason, _), do: reason
defp exit_reason(:error, reason, stack), do: {reason, stack}
defp exit_reason(:throw, value, stack), do: {{:nocatch, value}, stack}
@impl true
def handle_cast(_msg, state) do
{:noreply, [], state}
end
@impl true
def handle_info({:EXIT, pid, reason}, state) do
case maybe_restart_child(pid, reason, state) do
{:ok, state} -> {:noreply, [], state}
{:shutdown, state} -> {:stop, :shutdown, state}
end
end
def handle_info({:"$gen_restart", pid}, state) do
%{children: children, template: child, restarting: restarting} = state
state = %{state | restarting: restarting - 1}
case children do
%{^pid => restarting_args} ->
{:restarting, [producer | args]} = restarting_args
case restart_child(producer, pid, args, child, state) do
{:ok, state} ->
{:noreply, [], state}
{:shutdown, state} ->
{:stop, :shutdown, state}
end
# We may hit clause if we send $gen_restart and then
# someone calls terminate_child, removing the child.
%{} ->
{:noreply, [], state}
end
end
def handle_info(msg, state) do
:error_logger.error_msg(~c"ConsumerSupervisor received unexpected message: ~tp~n", [msg])
{:noreply, [], state}
end
@impl true
def code_change(_, %{mod: mod, args: args} = state, _) do
case mod.init(args) do
{:ok, children, opts} ->
case validate_specs(children) do
:ok ->
case init(state, children, opts) do
{:ok, state, _} -> {:ok, state}
{:error, message} -> {:error, {:bad_opts, message}}
end
{:error, message} ->
{:error, {:bad_specs, message}}
end
:ignore ->
{:ok, state}
error ->
error
end
end
@impl true
def terminate(_, %{children: children} = state) do
:ok = terminate_children(children, state)
end
defp terminate_children(children, %{template: template} = state) do
{_, _, restart, shutdown, _, _} = template
{pids, stacks} = monitor_children(children, restart)
size = map_size(pids)
stacks =
case shutdown do
:brutal_kill ->
for {pid, _} <- pids, do: Process.exit(pid, :kill)
wait_children(restart, shutdown, pids, size, nil, stacks)
:infinity ->
for {pid, _} <- pids, do: Process.exit(pid, :shutdown)
wait_children(restart, shutdown, pids, size, nil, stacks)
time ->
for {pid, _} <- pids, do: Process.exit(pid, :shutdown)
timer = :erlang.start_timer(time, self(), :kill)
wait_children(restart, shutdown, pids, size, timer, stacks)
end
for {pid, reason} <- stacks do
report_error(:shutdown_error, reason, pid, :undefined, template, state)
end
:ok
end
defp monitor_children(children, restart) do
Enum.reduce(children, {%{}, %{}}, fn
{_, {:restarting, _}}, {pids, stacks} ->
{pids, stacks}
{pid, _}, {pids, stacks} ->
case monitor_child(pid) do
:ok ->
{Map.put(pids, pid, true), stacks}
{:error, :normal} when restart != :permanent ->
{pids, stacks}
{:error, reason} ->
{pids, Map.put(stacks, pid, reason)}
end
end)
end
defp monitor_child(pid) do
ref = Process.monitor(pid)
Process.unlink(pid)
receive do
{:EXIT, ^pid, reason} ->
receive do
{:DOWN, ^ref, :process, ^pid, _} -> {:error, reason}
end
after
0 -> :ok
end
end
defp wait_children(_restart, _shutdown, _pids, 0, nil, stacks) do
stacks
end
defp wait_children(_restart, _shutdown, _pids, 0, timer, stacks) do
_ = :erlang.cancel_timer(timer)
receive do
{:timeout, ^timer, :kill} -> :ok
after
0 -> :ok
end
stacks
end
defp wait_children(restart, :brutal_kill, pids, size, timer, stacks) do
receive do
{:DOWN, _ref, :process, pid, :killed} ->
wait_children(restart, :brutal_kill, Map.delete(pids, pid), size - 1, timer, stacks)
{:DOWN, _ref, :process, pid, reason} ->
wait_children(
restart,
:brutal_kill,
Map.delete(pids, pid),
size - 1,
timer,
Map.put(stacks, pid, reason)
)
end
end
defp wait_children(restart, shutdown, pids, size, timer, stacks) do
receive do
{:DOWN, _ref, :process, pid, {:shutdown, _}} ->
wait_children(restart, shutdown, Map.delete(pids, pid), size - 1, timer, stacks)
{:DOWN, _ref, :process, pid, :shutdown} ->
wait_children(restart, shutdown, Map.delete(pids, pid), size - 1, timer, stacks)
{:DOWN, _ref, :process, pid, :normal} when restart != :permanent ->
wait_children(restart, shutdown, Map.delete(pids, pid), size - 1, timer, stacks)
{:DOWN, _ref, :process, pid, reason} ->
stacks = Map.put(stacks, pid, reason)
wait_children(restart, shutdown, Map.delete(pids, pid), size - 1, timer, stacks)
{:timeout, ^timer, :kill} ->
for {pid, _} <- pids, do: Process.exit(pid, :kill)
wait_children(restart, shutdown, pids, size, nil, stacks)
end
end
defp maybe_restart_child(pid, reason, state) do
%{children: children, template: child} = state
{_, _, restart, _, _, _} = child
case children do
%{^pid => [producer | args]} ->
maybe_restart_child(restart, reason, producer, pid, args, child, state)
%{} ->
{:ok, state}
end
end
defp maybe_restart_child(:permanent, reason, producer, pid, args, child, state) do
report_error(:child_terminated, reason, pid, args, child, state)
restart_child(producer, pid, args, child, state)
end
defp maybe_restart_child(_, :normal, producer, pid, _args, _child, state) do
{:ok, delete_child_and_maybe_ask(producer, pid, state)}
end
defp maybe_restart_child(_, :shutdown, producer, pid, _args, _child, state) do
{:ok, delete_child_and_maybe_ask(producer, pid, state)}
end
defp maybe_restart_child(_, {:shutdown, _}, producer, pid, _args, _child, state) do
{:ok, delete_child_and_maybe_ask(producer, pid, state)}
end
defp maybe_restart_child(:transient, reason, producer, pid, args, child, state) do
report_error(:child_terminated, reason, pid, args, child, state)
restart_child(producer, pid, args, child, state)
end
defp maybe_restart_child(:temporary, reason, producer, pid, args, child, state) do
report_error(:child_terminated, reason, pid, args, child, state)
{:ok, delete_child_and_maybe_ask(producer, pid, state)}
end
defp delete_child_and_maybe_ask(:dynamic, pid, %{children: children} = state) do
%{state | children: Map.delete(children, pid)}
end
defp delete_child_and_maybe_ask(ref, pid, %{children: children} = state) do
children = Map.delete(children, pid)
maybe_ask(ref, pid, 0, 1, children, state)
end
defp restart_child(producer, pid, args, child, state) do
case add_restart(state) do
{:ok, %{strategy: strategy} = state} ->
case restart_child(strategy, producer, pid, args, child, state) do
{:ok, state} ->
{:ok, state}
{:try_again, state} ->
send(self(), {:"$gen_restart", pid})
{:ok, state}
end
{:shutdown, state} ->
report_error(:shutdown, :reached_max_restart_intensity, pid, args, child, state)
{:shutdown, delete_child_and_maybe_ask(producer, pid, state)}
end
end
defp add_restart(state) do
%{max_seconds: max_seconds, max_restarts: max_restarts, restarts: restarts} = state
now = :erlang.monotonic_time(1)
restarts = add_restart([now | restarts], now, max_seconds)
state = %{state | restarts: restarts}
if length(restarts) <= max_restarts do
{:ok, state}
else
{:shutdown, state}
end
end
defp add_restart(restarts, now, period) do
for then <- restarts, now <= then + period, do: then
end
defp restart_child(:one_for_one, producer, current_pid, args, child, state) do
{_, {m, f, _}, restart, _, _, _} = child
case start_child(m, f, args) do
{:ok, pid, _} ->
state = %{state | children: Map.delete(state.children, current_pid)}
{:ok, save_child(restart, producer, pid, args, state)}
{:ok, pid} ->
state = %{state | children: Map.delete(state.children, current_pid)}
{:ok, save_child(restart, producer, pid, args, state)}
:ignore ->
{:ok, delete_child_and_maybe_ask(producer, current_pid, state)}
{:error, reason} ->
report_error(:start_error, reason, {:restarting, current_pid}, args, child, state)
state = restart_child(current_pid, state)
{:try_again, update_in(state.restarting, &(&1 + 1))}
end
end
defp restart_child(pid, %{children: children} = state) do
case children do
%{^pid => {:restarting, _}} ->
state
%{^pid => info} ->
%{state | children: Map.put(children, pid, {:restarting, info})}
end
end
defp report_error(error, reason, pid, args, child, %{name: name}) do
:error_logger.error_report(
:supervisor_report,
supervisor: name,
errorContext: error,
reason: reason,
offender: extract_child(pid, args, child)
)
end
defp extract_child(pid, args, {id, {m, f, _}, restart, shutdown, type, _}) do
[
pid: pid,
id: id,
mfargs: {m, f, args},
restart_type: restart,
shutdown: shutdown,
child_type: type
]
end
@impl true
def format_status(:terminate, [_pdict, state]) do
state
end
def format_status(_, [_pdict, %{mod: mod} = state]) do
[
data: [{~c"State", state}],
supervisor: [{~c"Callback", mod}]
]
end
defp normalize_template(%{id: id, start: {mod, _, _} = start} = child),
do: {
id,
start,
Map.get(child, :restart, :permanent),
Map.get(child, :shutdown, 5_000),
Map.get(child, :type, :worker),
Map.get(child, :modules, [mod])
}
defp normalize_template({_, _, _, _, _, _} = child), do: child
defp validate_template({_, _, :permanent, _, _, _}) do
error = """
a child specification with :restart set to :permanent \
is not supported in ConsumerSupervisor
Set the :restart option either to :temporary, so children \
spawned from events are never restarted, or :transient, \
so they are restarted only on abnormal exits
"""
{:error, error}
end
defp validate_template({_, _, _, _, _, _}) do
:ok
end
end
