Refactor conditional stages and add typespecs

Author: zorbash

.formatter.exs

@@ -10,7 +10,7 @@
       check: 2,
       link: 1,
       link: 2,
-      skip: 1,
+      skip: 2,
       instrument: 2,
       instrument: 3,
       send: 2
@@ -26,7 +26,7 @@
     check: 2,
     link: 1,
     link: 2,
-    skip: 1,
+    skip: 2,
     instrument: 2,
     instrument: 3,
     send: 2,

README.md

@@ -69,7 +69,6 @@ example above you need to add `use Opus.Pipeline` to turn a module into
 a pipeline. A pipeline module is a composition of stages executed in
 sequence.
 
-
 ## Stages
 
 There are a few different types of stages for different use-cases.
@@ -116,6 +115,50 @@ This stage is to link with another Opus.Pipeline module. It calls
 `call/1` for the provided module. If the module is not an
 `Opus.Pipeline` it is ignored.
 
+#### Skip
+
+The `skip` macro can be used for linked pipelines.
+A linked pipeline may act as a true bypass, based on a condition,
+expressed as either `:if` or `:unless`. When skipped, none of the stages
+are executed and it returns the input, to be used by any next stages of
+the caller pipeline. A very common use-case is illustrated in the following example:
+
+
+```elixir
+defmodule RetrieveCustomerInformation do
+  use Opus.Pipeline
+
+  check :valid_query?
+  link FetchFromCache,    if: :cacheable?
+  link FetchFromDatabase, if: :db_backed?
+  step :serialize
+end
+```
+
+With `skip` it can be written as:
+
+```elixir
+defmodule RetrieveCustomerInformation do
+  use Opus.Pipeline
+
+  check :valid_query?
+  link FetchFromCache
+  link FetchFromDatabase
+  step :serialize
+end
+```
+
+A linked pipeline becomes:
+
+```elixir
+defmodule FetchFromCache do
+  use Opus.Pipeline
+
+  skip :assert_suitable, if: :cacheable?
+  step :retrive_from_cache
+end
+```
+
 ### Available options
 
 The behaviour of each stage can be configured with any of the available
@@ -235,7 +278,7 @@ defmodule CustomInstrumentation do
   def instrument(:pipeline_started, %{pipeline: ArithmeticPipeline}, %{input: input}) do
     # publish the metrics to specific backend
   end
-  
+
   def instrument(:before_stage, %{stage: %{pipeline: pipeline}}, %{input: input}) do
     # publish the metrics to specific backend
   end
@@ -254,6 +297,11 @@ end
 
 You may choose to provide some common options to all the stages of a pipeline.
 
+* `:raise`: A list of exceptions to not rescue. When set to `true`, Opus
+    does not handle any exceptions. Defaults to `false` which converts all exceptions
+    to `{:error, %Opus.PipelineError{}}` values halting the pipeline.
+* `:instrument?`: A boolean which defaults to `true`. Set to `false` to
+  skip instrumentation for a module.
 
 ```elixir
 defmodule ArithmeticPipeline do

lib/opus/pipeline.ex

@@ -12,10 +12,10 @@ defmodule Opus.Pipeline do
         step :double, with: & &1 * 2
       end
 
-  The pipeline can be run calling a `call/1` function which is defined by using Opus.Pipeline.
+  The pipeline can be run calling a `call/1` function which is defined by `Opus.Pipeline`.
   Pipelines are intended to have a single parameter and always return a tagged tuple `{:ok, value} | {:error, error}`.
   A stage returning `{:error, error}` halts the pipeline. The error value is an `Opus.PipelineError` struct which
-  contains useful information to detect where was the error caused and why.
+  contains useful information to detect where the error was caused and why.
 
   ## Exception Handling
 
@@ -40,6 +40,8 @@ defmodule Opus.Pipeline do
   CreateUserPipeline.call(params, except: :send_notification)
   ```
   """
+  @type opts :: [only: [atom], except: [atom]]
+  @type result :: {:ok, any} | {:error, Opus.PipelineError.t()}
 
   defmacro __using__(opts) do
     quote location: :keep do
@@ -63,6 +65,8 @@ defmodule Opus.Pipeline do
       @doc false
       def pipeline?, do: true
 
+      @doc "The entrypoint function of the pipeline"
+      @spec call(any, Opus.Pipeline.opts()) :: Opus.Pipeline.result()
       def call(input, opts \\ %{}) do
         instrument? = Pipeline._opus_opts()[:instrument?]
 
@@ -97,6 +101,7 @@ defmodule Opus.Pipeline do
         end
       end
 
+      @doc false
       def _opus_opts, do: @opus_opts
 
       defp run_instrumented({type, name, opts} = stage, %{time: acc_time, input: input}) do
@@ -112,9 +117,15 @@ defmodule Opus.Pipeline do
           )
 
         case instrumented_return do
+          {status, %{time: time, input: :pipeline_skipped}} ->
+            {status, %{time: acc_time + time, input: input}}
+
           {status, %{time: time, input: new_input}} ->
             {status, %{time: acc_time + time, input: new_input}}
 
+          {status, :pipeline_skipped} ->
+            {status, %{time: acc_time + 0, input: input}}
+
           {status, new_input} ->
             {status, %{time: acc_time + 0, input: new_input}}
         end

lib/opus/pipeline/registration.ex

@@ -45,15 +45,38 @@ defmodule Opus.Pipeline.Registration do
 
   def maybe_define_callbacks(stage_id, name, opts) do
     [
-      define_callback(:if, stage_id, name, Access.get(opts, :if)),
-      define_callback(:unless, stage_id, name, Access.get(opts, :unless)),
+      define_callback(:conditional, stage_id, name, ensure_valid_conditional!(opts)),
       define_callback(:with, stage_id, name, Access.get(opts, :with)),
       define_callback(:retry_backoff, stage_id, name, Access.get(opts, :retry_backoff))
     ]
   end
 
+  def ensure_valid_conditional!(opts) do
+    if_cond = Access.get(opts, :if)
+    unless_cond = Access.get(opts, :unless)
+
+    if if_cond && unless_cond do
+      raise CompileError,
+        file: __ENV__.file,
+        line: __ENV__.line,
+        description:
+          "Invalid stage conditional. For each stage you may define either an :if or an :unless option. Not both"
+    end
+
+    if_cond || unless_cond
+  end
+
   def normalize_opts(opts, id, callbacks) do
     callback_types = for %{stage_id: ^id, type: type} <- callbacks, do: type
-    Keyword.merge(opts, for(t <- callback_types, do: {t, :anonymous}))
+
+    for {k, v} <- opts, into: [] do
+      anonymous? = k in callback_types || (k in [:if, :unless] && :conditional in callback_types)
+      callback = if anonymous?, do: :anonymous, else: v
+
+      case k do
+        term when term in [:if, :unless] -> {:conditional, {term, callback}}
+        _ -> {k, callback}
+      end
+    end
   end
 end

lib/opus/pipeline/stage.ex

@@ -15,69 +15,68 @@ defmodule Opus.Pipeline.Stage do
 
   alias Opus.{Safe, PipelineError}
 
-  def maybe_run({module, type, name, %{if: :anonymous, stage_id: id} = opts}, input) do
-    callback = (module._opus_callbacks[id] |> Enum.find(fn %{type: t} -> t == :if end)).name
-    maybe_run({module, type, name, %{opts | if: {module, callback, [input]}}}, input)
-  end
-
-  def maybe_run({module, type, name, %{if: fun} = opts} = _stage, input)
-      when is_atom(fun),
-      do: maybe_run({module, type, name, %{opts | if: {module, fun, [input]}}}, input)
-
-  def maybe_run({module, :skip, name, %{if: {_m, _f, _a} = condition}}, input) do
-    case Safe.apply(condition) do
-      true ->
-        module.instrument(:pipeline_skipped, %{stage: %{pipeline: module, name: name}}, %{
-          stage: name,
-          input: input
-        })
-
-        # Stop the pipeline execution
-        :pipeline_skipped
+  @doc false
+  def maybe_run(
+        {module, type, name, %{conditional: {cond_type, :anonymous}, stage_id: id} = opts},
+        input
+      ) do
+    callback =
+      (module._opus_callbacks[id] |> Enum.find(fn %{type: t} -> t == :conditional end)).name
 
-      _ ->
-        nil
-    end
+    maybe_run(
+      {module, type, name, %{opts | conditional: {cond_type, {module, callback, [input]}}}},
+      input
+    )
   end
 
-  def maybe_run({module, _type, name, %{if: {_m, _f, _a} = condition} = opts} = stage, input) do
-    case Safe.apply(condition) do
-      true ->
-        with_retries({module, opts}, fn -> do_run(stage, input) end)
-
-      _ ->
-        module.instrument(:stage_skipped, %{stage: %{pipeline: module, name: name}}, %{
-          stage: name,
-          input: input
-        })
+  def maybe_run(
+        {module, :skip, name, %{conditional: {cond_type, {_m, _f, _a} = condition}}},
+        input
+      ) do
+    if eval_condition(cond_type, condition) do
+      module.instrument(:pipeline_skipped, %{stage: %{pipeline: module, name: name}}, %{
+        stage: name,
+        input: input
+      })
 
-        # Ignore this stage
-        :stage_skipped
+      # Stop the pipeline execution
+      :pipeline_skipped
+    else
+      nil
     end
   end
 
-  def maybe_run({module, type, name, %{unless: :anonymous, stage_id: id} = opts}, input) do
-    callback = (module._opus_callbacks[id] |> Enum.find(fn %{type: t} -> t == :unless end)).name
-    maybe_run({module, type, name, %{opts | unless: {module, callback, [input]}}}, input)
+  def maybe_run(
+        {
+          module,
+          type,
+          name,
+          %{conditional: {cond_type, fun}} = opts
+        } = _stage,
+        input
+      )
+      when is_atom(fun) do
+    maybe_run(
+      {module, type, name, %{opts | conditional: {cond_type, {module, fun, [input]}}}},
+      input
+    )
   end
 
-  def maybe_run({module, type, name, %{unless: fun} = opts} = _stage, input)
-      when is_atom(fun),
-      do: maybe_run({module, type, name, %{opts | unless: {module, fun, [input]}}}, input)
-
-  def maybe_run({module, _type, name, %{unless: {_m, _f, _a} = condition} = opts} = stage, input) do
-    case Safe.apply(condition) do
-      false ->
-        with_retries({module, opts}, fn -> do_run(stage, input) end)
-
-      _ ->
-        module.instrument(:stage_skipped, %{stage: %{pipeline: module, name: name}}, %{
-          stage: name,
-          input: input
-        })
+  def maybe_run(
+        {module, _type, name, %{conditional: {cond_type, {_m, _f, _a} = condition}} = opts} =
+          stage,
+        input
+      ) do
+    if eval_condition(cond_type, condition) do
+      with_retries({module, opts}, fn -> do_run(stage, input) end)
+    else
+      module.instrument(:stage_skipped, %{stage: %{pipeline: module, name: name}}, %{
+        stage: name,
+        input: input
+      })
 
-        # Ignore this stage
-        :stage_skipped
+      # Ignore this stage
+      :stage_skipped
     end
   end
 
@@ -201,4 +200,12 @@ defmodule Opus.Pipeline.Stage do
 
   defp do_run({module, type, name, %{} = opts}, input),
     do: do_run({module, type, name, Map.merge(opts, %{with: {module, name, [input]}})}, input)
+
+  defp eval_condition(cond_type, condition) do
+    case {cond_type, Safe.apply(condition)} do
+      {:if, true} -> true
+      {:unless, false} -> true
+      _ -> false
+    end
+  end
 end

lib/opus/pipeline/stage/skip.ex

@@ -11,7 +11,7 @@ defmodule Opus.Pipeline.Stage.Skip do
   defmodule CreateUserPipeline do
     use Opus.Pipeline
 
-    skip if: :user_exists?
+    skip :prevent_duplicates, if: :user_exists?
     step :persist_user
   end
   ```
@@ -24,9 +24,9 @@ defmodule Opus.Pipeline.Stage.Skip do
 
   @behaviour Stage
 
-  def run({module, type, [if: func], opts}, input) do
-    case Stage.maybe_run({module, type, nil, opts |> put_in([:if], func)}, input) do
-      :pipeline_skipped -> {:halt, :skipped}
+  def run(stage, input) do
+    case stage |> Stage.maybe_run(input) do
+      :pipeline_skipped -> {:halt, :pipeline_skipped}
       _ -> {:cont, input}
     end
   end

lib/opus/pipeline_error.ex

@@ -3,6 +3,14 @@ defmodule Opus.PipelineError do
   Error struct capturing useful information to detect where an error was caused and why.
   """
 
+  @type t :: %__MODULE__{
+          error: struct,
+          input: any,
+          pipeline: module,
+          stage: atom,
+          stacktrace: [{module, atom, non_neg_integer, keyword}]
+        }
+
   defexception [:error, :pipeline, :stage, :input, :stacktrace]
 
   def message(%{error: error, pipeline: pipeline, stage: stage, input: input, stacktrace: trace}) do

mix.exs

@@ -41,6 +41,7 @@ defmodule Opus.Mixfile do
       {:retry, "~> 0.8"},
       {:credo, "~> 0.8.10", only: [:dev, :test], runtime: false},
       {:ex_doc, "~> 0.18.0", only: :dev, runtime: false},
+      {:dialyxir, "~> 1.0.0-rc.3", only: [:dev, :test], runtime: false},
       {:excoveralls, "~> 0.8", only: :test}
     ]
   end

mix.lock

@@ -2,7 +2,9 @@
   "bunt": {:hex, :bunt, "0.2.0", "951c6e801e8b1d2cbe58ebbd3e616a869061ddadcc4863d0a2182541acae9a38", [:mix], [], "hexpm"},
   "certifi": {:hex, :certifi, "2.5.1", "867ce347f7c7d78563450a18a6a28a8090331e77fa02380b4a21962a65d36ee5", [:rebar3], [{:parse_trans, "~>3.3", [hex: :parse_trans, repo: "hexpm", optional: false]}], "hexpm"},
   "credo": {:hex, :credo, "0.8.10", "261862bb7363247762e1063713bb85df2bbd84af8d8610d1272cd9c1943bba63", [:mix], [{:bunt, "~> 0.2.0", [hex: :bunt, repo: "hexpm", optional: false]}], "hexpm"},
+  "dialyxir": {:hex, :dialyxir, "1.0.0-rc.6", "78e97d9c0ff1b5521dd68041193891aebebce52fc3b93463c0a6806874557d7d", [:mix], [{:erlex, "~> 0.2.1", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm"},
   "earmark": {:hex, :earmark, "1.3.1", "73812f447f7a42358d3ba79283cfa3075a7580a3a2ed457616d6517ac3738cb9", [:mix], [], "hexpm"},
+  "erlex": {:hex, :erlex, "0.2.1", "cee02918660807cbba9a7229cae9b42d1c6143b768c781fa6cee1eaf03ad860b", [:mix], [], "hexpm"},
   "ex_doc": {:hex, :ex_doc, "0.18.4", "4406b8891cecf1352f49975c6d554e62e4341ceb41b9338949077b0d4a97b949", [:mix], [{:earmark, "~> 1.1", [hex: :earmark, repo: "hexpm", optional: false]}], "hexpm"},
   "excoveralls": {:hex, :excoveralls, "0.10.5", "7c912c4ec0715a6013647d835c87cde8154855b9b84e256bc7a63858d5f284e3", [:mix], [{:hackney, "~> 1.13", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm"},
   "hackney": {:hex, :hackney, "1.15.1", "9f8f471c844b8ce395f7b6d8398139e26ddca9ebc171a8b91342ee15a19963f4", [:rebar3], [{:certifi, "2.5.1", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "6.0.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "1.0.1", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~>1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "1.1.4", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm"},

test/opus/pipeline/stage/option/if_test.exs

@@ -1,7 +1,7 @@
 defmodule Opus.Pipeline.Option.IfTest do
   use ExUnit.Case
 
-  describe "as atom - when the condition is truth" do
+  describe "as atom - when the condition is true" do
     defmodule PipelineWithTruthyAtomIf do
       use Opus.Pipeline