Current section

Files

Jump to
reactor lib reactor builder step.ex
Raw

lib/reactor/builder/step.ex

defmodule Reactor.Builder.Step do
@moduledoc """
Handle building and adding steps to Reactors for the builder.
You should not use this module directly, but instead use
`Reactor.Builder.new_step/4` and `Reactor.Builder.add_step/5`.
"""
alias Reactor.{Argument, Builder, Guard, Step, Template}
import Reactor.Argument, only: :macros
import Reactor.Utils
@option_schema [
async?: [
type: {:or, [:boolean, {:mfa_or_fun, 1}]},
default: true,
required: false,
doc: "Allow the step to be run asynchronously?"
],
max_retries: [
type: {:or, [:non_neg_integer, {:literal, :infinity}]},
default: 100,
required: false,
doc: "The maximum number of times the step can ask to be retried"
],
transform: [
type:
{:or,
[
nil,
{:mfa_or_fun, 1},
{:tuple, [:module, :non_empty_keyword_list]}
]},
required: false,
doc: "A function which can modify all incoming arguments"
],
context: [
type: :map,
required: false,
default: %{},
doc: "Context which will be merged with the reactor context when calling this step"
],
description: [
type: {:or, [nil, :string]},
required: false,
doc: "An optional description for the step"
],
guards: [
type: {:list, {:protocol, Reactor.Guard.Build}},
required: false,
default: [],
doc: "Optional guards to add to the step"
],
ref: [
type: {:in, [:step_name, :make_ref]},
required: false,
default: :make_ref,
doc: "What sort of step reference to generate"
]
]
@doc """
Build and add a new step to a Reactor.
## Arguments
* `reactor` - An existing Reactor struct to add the step to.
* `name` - The proposed name of the new step.
* `impl` - A module implementing the `Reactor.Step` behaviour (or a tuple
containing the module and options).
* `arguments` - A list of `Reactor.Argument` structs or shorthand keyword
lists.
## Options
#{Spark.Options.docs(@option_schema)}
"""
@doc spark_opts: [{5, @option_schema}]
@spec add_step(
Reactor.t(),
any,
Builder.impl(),
[Builder.step_argument()],
Builder.step_options()
) :: {:ok, Reactor.t()} | {:error, any}
def add_step(reactor, name, impl, arguments, options) do
with {:ok, options} <- Spark.Options.validate(options, @option_schema),
{:ok, arguments} <- build_arguments(arguments),
{:ok, arguments} <- maybe_rewrite_input_arguments(reactor, arguments),
{:ok, guards} <- build_guards(options[:guards]),
:ok <- assert_is_step_impl(impl),
{:ok, {arguments, argument_transform_steps}} <-
build_argument_transform_steps(arguments, name),
{:ok, arguments, step_transform_step} <-
maybe_build_step_transform_all_step(arguments, name, options[:transform]) do
context =
if step_transform_step do
deep_merge(options[:context], %{private: %{replace_arguments: :value}})
else
options[:context]
end
steps =
[
%Step{
arguments: arguments,
async?: options[:async?],
context: context,
description: options[:description],
guards: guards,
impl: impl,
name: name,
max_retries: options[:max_retries]
}
]
|> Enum.concat(argument_transform_steps)
|> maybe_append(step_transform_step)
|> Enum.concat(reactor.steps)
|> Enum.map(&set_ref(&1, options[:ref]))
{:ok, %{reactor | steps: steps}}
end
end
defp set_ref(step, :step_name), do: %{step | ref: step.name}
defp set_ref(step, _), do: %{step | ref: make_ref()}
@doc """
Dynamically build a new step for later use.
You're most likely to use this when dynamically returning new steps from an
existing step.
## Arguments
* `name` - The name of the new step.
* `impl` - A module implementing the `Reactor.Step` behaviour (or a tuple
containing the module and options).
* `arguments` - A list of `Reactor.Argument` structs or shorthand keyword
lists.
## Options
#{Spark.Options.docs(@option_schema)}
"""
@doc spark_opts: [{5, @option_schema}]
@spec new_step(
any,
Builder.impl(),
[Builder.step_argument()],
Builder.step_options()
) ::
{:ok, Step.t()} | {:error, any}
def new_step(name, impl, arguments, options) do
with {:ok, options} <- Spark.Options.validate(options, @option_schema),
{:ok, arguments} <- build_arguments(arguments),
:ok <- assert_no_argument_transforms(arguments),
{:ok, guards} <- build_guards(options[:guards]),
:ok <- assert_is_step_impl(impl),
:ok <- validate_no_transform_option(options) do
step = %Step{
arguments: arguments,
async?: options[:async?],
context: options[:context],
description: options[:description],
guards: guards,
impl: impl,
name: name,
max_retries: options[:max_retries]
}
{:ok, step}
end
end
defp validate_no_transform_option(options) do
if Keyword.has_key?(options, :transform) do
{:error,
argument_error(:options, "Adding transforms to dynamic steps is not supported.", options)}
else
:ok
end
end
defp build_arguments(arguments) do
arguments
|> map_while_ok(&Argument.Build.build/1)
|> and_then(&{:ok, List.flatten(&1)})
end
defp maybe_rewrite_input_arguments(reactor, arguments) do
existing_step_names = MapSet.new(reactor.steps, & &1.name)
map_while_ok(arguments, fn
argument when is_from_input(argument) ->
potential_rewrite_step_name = {:__reactor__, :transform, :input, argument.source.name}
if MapSet.member?(existing_step_names, potential_rewrite_step_name) do
{:ok, Argument.from_result(argument.name, potential_rewrite_step_name)}
else
{:ok, argument}
end
argument
when is_from_result(argument) or is_from_value(argument) or is_from_element(argument) ->
{:ok, argument}
end)
end
defp build_guards(guards) do
guards
|> map_while_ok(&Guard.Build.build/1)
|> and_then(&{:ok, List.flatten(&1)})
end
defp assert_is_step_impl({impl, opts}) when is_list(opts), do: assert_is_step_impl(impl)
defp assert_is_step_impl(impl) when is_atom(impl) do
if Spark.implements_behaviour?(impl, Step) do
:ok
else
{:error,
argument_error(:impl, "Module does not implement the `Reactor.Step` behaviour.", impl)}
end
end
defp build_argument_transform_steps(arguments, step_name) do
arguments
|> reduce_while_ok({[], []}, fn
argument, {arguments, steps} when is_from_input(argument) and has_transform(argument) ->
transform_step_name = {:__reactor__, :transform, argument.name, step_name}
step =
build_transform_step(
argument.source,
transform_step_name,
argument.transform
)
argument = %Argument{
name: argument.name,
source: %Template.Result{name: transform_step_name}
}
{:ok, {[argument | arguments], [%{step | transform: nil} | steps]}}
argument, {arguments, steps} when is_from_result(argument) and has_transform(argument) ->
transform_step_name = {:__reactor__, :transform, argument.name, step_name}
step =
build_transform_step(
argument.source,
transform_step_name,
argument.transform
)
argument = %Argument{
name: argument.name,
source: %Template.Result{name: transform_step_name}
}
{:ok, {[argument | arguments], [%{step | transform: nil} | steps]}}
argument, {arguments, steps} ->
{:ok, {[argument | arguments], steps}}
end)
end
defp maybe_build_step_transform_all_step(arguments, _name, nil), do: {:ok, arguments, nil}
defp maybe_build_step_transform_all_step(arguments, name, transform)
when is_function(transform, 1),
do:
maybe_build_step_transform_all_step(arguments, name, {Step.TransformAll, fun: transform})
defp maybe_build_step_transform_all_step(arguments, name, transform) do
step = %Step{
arguments: arguments,
async?: false,
impl: transform,
name: {:__reactor__, :transform, name},
max_retries: 0
}
{:ok, [Argument.from_result(:value, step.name)], step}
end
defp build_transform_step(argument_source, step_name, transform) when is_function(transform, 1),
do: build_transform_step(argument_source, step_name, {Step.Transform, fun: transform})
defp build_transform_step(argument_source, step_name, transform)
when tuple_size(transform) == 2 and is_atom(elem(transform, 0)) and
is_list(elem(transform, 1)) do
%Step{
arguments: [
%Argument{
name: :value,
source: argument_source
}
],
async?: false,
impl: transform,
name: step_name,
max_retries: 0
}
end
defp assert_no_argument_transforms(arguments) do
arguments
|> Enum.reject(&is_nil(&1.transform))
|> case do
[] ->
:ok
[argument] ->
{:error,
argument_error(
:arguments,
"Argument `#{argument.name}` has a transform attached.",
argument
)}
arguments ->
sentence = sentence(arguments, &"`#{&1.name}`", ", ", " and ")
{:error,
argument_error(:arguments, "Arguments #{sentence} have transforms attached.", arguments)}
end
end
end