# SPDX-FileCopyrightText: 2025 reactor_process contributors # # SPDX-License-Identifier: MIT defmodule Reactor.Process.Step.StartLink do @arg_schema Spark.Options.new!( child_spec: [ type: {:or, [{:tuple, [:module, :keyword_list]}, :module]}, required: true, doc: "The child spec" ] ) @opt_schema Spark.Options.new!( fail_on_already_started?: [ type: :boolean, required: false, default: true, doc: "Whether the step should fail if the start function returns an already started error" ], fail_on_ignore?: [ type: :boolean, required: false, default: true, doc: "Whether the step should fail if the start function returns `:ignore`" ], terminate_on_undo?: [ type: :boolean, required: false, default: true, doc: "Whether to terminate the started process when the Reactor is undoing changes" ], termination_reason: [ type: :any, required: false, default: :kill, doc: "The reason to give to the process when terminating it" ], termination_timeout: [ type: :timeout, required: false, default: 5_000, doc: "How long to wait for a process to terminate" ] ) @moduledoc """ A Reactor step which starts a process via it's child spec and links it to the process which is running the Reactor. > #### Warning {: .warning} > If you are building your Reactor directly (rather than via the DSL) then > this step must be added with the `async?: false` option otherwise the step > will fail. The `child_spec` argument expects a `t:Supervisor.module_spec()`. The step will then use the `start` MFA returned by the module's `child_spec/1` function to start the child process. ## Arguments #{Spark.Options.docs(@arg_schema)} ## Options #{Spark.Options.docs(@opt_schema)} """ use Reactor.Step alias Reactor.Process.Errors.MissingMiddlewareError import Reactor.Process.Utils @doc false @impl true def run(arguments, %{Reactor.Process.Middleware => %{pid: pid}} = _context, options) when pid == self() do with {:ok, arguments} <- Spark.Options.validate(Enum.to_list(arguments), @arg_schema), {:ok, options} <- Spark.Options.validate(options, @opt_schema), {:ok, child_spec} <- child_spec(arguments[:child_spec]) do start_child(child_spec, options) end end def run(_arguments, context, _options) do {:error, MissingMiddlewareError.exception( step: context.current_step, message: """ This step is not running in the same process as the Reactor, which means it has been run without the `async?: false` option. """ )} end @doc false @impl true def async?(_), do: false @doc false @impl true def can?(%{impl: {_, options}}, :undo), do: Keyword.get(options, :terminate_on_undo?, true) def can?(_, :undo), do: false def can?(step, capability), do: super(step, capability) @doc false @impl true def undo(process, _, context, options) do with {:ok, options} <- Spark.Options.validate(options, @opt_schema) do if Keyword.get(options, :terminate_on_undo?, true) do terminate( process, options[:termination_reason], options[:termination_timeout], context.current_step ) else :ok end end end defp start_child(%{start: {module, function, args}}, options) do fail_on_already_started? = options[:fail_on_already_started?] fail_on_ignore? = options[:fail_on_ignore?] case apply(module, function, args) do {:ok, pid} -> Process.link(pid) {:ok, pid} :ignore when fail_on_ignore? == true -> {:error, "Child process returned `:ignore`"} :ignore -> {:ok, :ignore} {:error, {:already_started, pid}} when fail_on_already_started? == true -> {:error, {:already_started, pid}} {:error, {:already_started, pid}} -> Process.link(pid) {:ok, pid} {:error, reason} -> {:error, reason} end end end