diff --git a/lib/sage.ex b/lib/sage.ex index 5b5dae2..6c7438d 100644 --- a/lib/sage.ex +++ b/lib/sage.ex @@ -163,6 +163,16 @@ defmodule Sage do defguardp is_transaction(value) when is_function(value, 2) or is_mfa(value) + @typedoc """ + Just like a transaction, although it receives extra argument of the name of the previous transaction + """ + @type intermediate_transaction :: + (effects_so_far :: effects(), attrs :: any(), previous_stage :: stage_name() -> + {:ok | :error | :abort, any()}) + | mfa() + + defguardp is_intermediate_transaction(value) when is_function(value, 3) or is_mfa(value) + @typedoc """ Tracer callback, can be a module that implements `Sage.Tracer` behaviour, an anonymous function, or an `{module, function, [arguments]}` tuple. @@ -437,6 +447,52 @@ defmodule Sage do def run_async(sage, name, transaction, compensation, opts \\ []), do: add_stage(sage, name, build_operation!(:run_async, transaction, compensation, opts)) + @doc """ + For a given Sage S with transactions `:t1` -> `:t2` -> `:t3`, a call to `interleave(S, :name, f)` + will yield a saga with transactions `:t1` -> `{:interleave, :name, 1}` -> `:t2` -> `{:interleave, :name, 2}` + -> `:t3` -> `{:interleave, :name, 3}`. + + This can be useful if you are trying to do a long computation and want to do something with + the intermediate results, such as logging or persistence. + + Note: + - This isn't strict interleaving because a transaction is still appended at the end. + - Calling this function twice with the same name will give a `Sage.DuplicateStageError`, but + calling the function twice with a different name will compound its effects. + - Calling this function before the end of your saga definition will mean any stages you add after the + call to `interleave` will not have the intermediate stages added after them + """ + @spec interleave( + sage :: t(), + name :: stage_name(), + intermediate_transaction :: intermediate_transaction(), + compensation :: compensation() + ) :: t() + def interleave(sage, name, intermediate_transaction, compensation \\ :noop) + when is_intermediate_transaction(intermediate_transaction) and is_compensation(compensation) do + new_stages = + sage.stages + |> Enum.reverse() + |> Enum.with_index(1) + |> Enum.flat_map(fn {stage, index} -> + name = {:interleave, name, index} + {stage_name, _} = stage + + transaction = + case intermediate_transaction do + {m, f, a} -> + {m, f, [stage_name | a]} + + _ -> + &intermediate_transaction.(&1, &2, stage_name) + end + + [stage, {name, build_operation!(:run, transaction, compensation)}] + end) + + add_stages(%{sage | stage_names: MapSet.new(), stages: []}, new_stages) + end + @doc """ Executes a Sage. @@ -497,16 +553,39 @@ defmodule Sage do end defp add_stage(sage, name, operation) do + add_stages(sage, [{name, operation}]) + end + + defp add_stages(sage, name_operation_pairs) do %{stages: stages, stage_names: names} = sage - if MapSet.member?(names, name) do - raise Sage.DuplicateStageError, sage: sage, name: name - else - %{ - sage - | stages: [{name, operation} | stages], - stage_names: MapSet.put(names, name) - } + {names_to_add_list, _operations} = + Enum.unzip(name_operation_pairs) + + names_to_add_set = MapSet.new(names_to_add_list) + + duplicates = + names_to_add_set + |> MapSet.intersection(names) + |> MapSet.to_list() + + cond do + # There is a duplicate between what was existing in the struct beforehand and + # what was passed to this function + !Enum.empty?(duplicates) -> + raise Sage.DuplicateStageError, sage: sage, name: hd(duplicates) + + # There was a duplicate within name_operation_pairs + length(names_to_add_list) != MapSet.size(names_to_add_set) -> + duplicates = names_to_add_list -- MapSet.to_list(names_to_add_set) + raise Sage.DuplicateStageError, sage: sage, name: hd(duplicates) + + true -> + %{ + sage + | stages: Enum.reverse(name_operation_pairs) ++ stages, + stage_names: MapSet.union(names, names_to_add_set) + } end end diff --git a/test/sage_test.exs b/test/sage_test.exs index c4c09ad..a4fb092 100644 --- a/test/sage_test.exs +++ b/test/sage_test.exs @@ -254,6 +254,92 @@ defmodule SageTest do end end + describe "interleaves/3" do + test "adds a step between every transaction" do + sage = + new() + |> run(:t1, transaction(:t1)) + |> run(:t2, transaction(:t2)) + |> run_async(:t_async, transaction(:t_async), :noop) + |> run(:t3, transaction(:t3)) + |> interleave(:i, fn _effects, _args, previous_stage_name -> {:ok, previous_stage_name} end) + + assert [ + {{:interleave, :i, 4}, _}, + {:t3, _}, + {{:interleave, :i, 3}, _}, + {:t_async, _}, + {{:interleave, :i, 2}, _}, + {:t2, _}, + {{:interleave, :i, 1}, _}, + t1: _ + ] = sage.stages + + assert {:ok, _, + %{ + {:interleave, :i, 4} => :t3, + {:interleave, :i, 3} => :t_async, + {:interleave, :i, 2} => :t2, + {:interleave, :i, 1} => :t1 + }} = execute(sage) + end + + test "adds nothing if there are no transactions" do + sage = interleave(new(), :i, fn _effects, _args, _previous_stage_name -> :ok end) + + assert sage.stages == [] + end + + test "adds a transaction at the end if there is one transaction" do + assert [{{:interleave, :i, 1}, _}, t1: _] = + new() + |> run(:t1, transaction(:t1)) + |> interleave(:i, fn _effects, _args, _previous_stage_name -> :ok end) + |> Map.get(:stages) + end + + test "works with mfa" do + sage = + new() + |> run(:t1, transaction(:t1)) + |> run(:t2, transaction(:t2)) + |> run(:t3, transaction(:t3)) + |> interleave(:i, {TestIntermediateTransactionHandler, :intermediate_transaction_handler, [:foo]}) + + assert {:ok, _, + %{ + {:interleave, :i, 3} => {:t3, :foo}, + {:interleave, :i, 2} => {:t2, :foo}, + {:interleave, :i, 1} => {:t1, :foo} + }} = execute(sage) + end + + test "can run a compensations" do + new() + |> run(:t1, transaction(:t1)) + |> run(:t2, transaction(:t2)) + |> run(:t3, transaction_with_error(:t3)) + |> interleave(:i, fn _effects, _args, _previous_stage_name -> {:ok, nil} end, fn _errored_effect, + _effects_so_far, + _attrs -> + send(self(), :compensating) + :ok + end) + |> execute() + + for _ <- 1..2, do: assert_received(:compensating) + end + + test "errors if used more than once" do + assert_raise Sage.DuplicateStageError, fn -> + new() + |> run(:t1, transaction(:t1)) + |> interleave(:i, fn _effects, _args, _previous_stage_name -> :ok end) + |> interleave(:i, fn _effects, _args, _previous_stage_name -> :ok end) + end + end + end + def dummy_transaction_for_mfa(_effects_so_far, _opts), do: raise("Not implemented") def dummy_compensation_for_mfa(_effect_to_compensate, _opts), do: raise("Not implemented") def dummy_final_cb(_status, _opts, _return), do: raise("Not implemented") diff --git a/test/support/test_intermediate_transaction_handler.ex b/test/support/test_intermediate_transaction_handler.ex new file mode 100644 index 0000000..a5d3906 --- /dev/null +++ b/test/support/test_intermediate_transaction_handler.ex @@ -0,0 +1,5 @@ +defmodule TestIntermediateTransactionHandler do + def intermediate_transaction_handler(_effects, _args, previous_stage_name, something_else) do + {:ok, {previous_stage_name, something_else}} + end +end