Hello,

What tool/template did you use to create the flowcharts on the Pipeline Execution wiki page? They look great.

Thanks,
Jeremy

defmodule Workflow do
  use Opus.Pipeline, strict: true

  # Here `strict: true` would require adherence to the type specs below
  @type strict_input :: map()
  @type strict_result ::
          :ok
          | {:ok, map}
          | {:error, any}
          | {:error, step :: atom, error :: any(), results_so_far :: map()}

  step :lookup_member 
  check :is_member_ready?

  matching %{member: %Member{status: "NO_GOOD"} = member} do
    tee(:send_email, with: fn -> Email.build(member, "not_ready.html") end)
  end

  matching %{member: %Member{status: "GOOD"} = member, payload: %{code: code}} do
    step :give_vip_status
    tee :expire_partner_code, with: fn -> Partner.expire_code(code) end

    matching %{member: %Member{status: "GREAT"} = member} do
      tee(:send_email, with: fn -> Email.build(member, "very_ready.html") end)
    end
  end

  defp lookup_member(%{member_id: id}) do
    {:ok, %{member: %Member{id: id}}}
  end

  defp give_vip_status(%{member: %Member{status: "Good"} = member}}) do
    {:ok, %{member | status: "Great"}}
  end
end

Explanation:

Thread a Kernel.match/1 call through children of the :do block's :if clauses. Which would have the benefit of avoiding the need to name the eval step

Design Concerns:

We would need a strict mode to ensure only patching the pipeline with maps from ok tuples

If there were some sort of strict mode, would it be better to use:

1. An Ecto.Multi style pipeline where each step puts its key and {:ok, result} into the map
2. An Exunit.Callbacks.setup/(1/2) style pipeline, which is just a folding map merge
3. A Plug.Conn style pipeline struct with nested :assigns and :results maps

PS

Could also name this bind, bind_match, with_match or otherwise.

Many times, the general behavior of a step is quite repetitive, so it would be nice to be able to reuse a single function of arity >1, and pass the extra arg(s) to the step itself.

If you think this could be interesting, I can provide a pull request.

This is an example:

defmodule Mapper do
  @moduledoc """
  Converts a flat structure from CSV into deeply nested fields of MyStruct.
  Empty fields are skipped.
  """

  use Opus.Pipeline

  step :parse
  step :nest, with: &%{source: &1}
  step :create_output, with: &Map.put(&1, :output, %MyStruct{})
  step :copy, args: ["some_col", [:some, :nested, :path]]
  step :copy, args: ["some_col_2", [:other, :nested, :path]]
  step :copy, args: ["some_col_3", [:again, :nested, :path]]
  # possibly many more lines like this
  step :extract_output, with: &get_in(&1, [:output])

  def parse(text) do
    [headers | rows] = NimbleCSV.RFC4180.parse_string(data, skip_headers: false)
    Enum.map(rows, fn row -> headers |> Enum.zip(row) |> Map.new() end)
  end

  def copy(input, [from, to]) do
    case get_in(input, [:source | List.wrap(from)]) do
      nil -> input
      "" -> input
      value -> put_in(input, [:output | List.wrap(to)], value)
    end
  end
end

On Elixir 1.8:

==> opus
Compiling 13 files (.ex)
warning: System.stacktrace/0 outside of rescue/catch clauses is deprecated. If you want to support only Elixir v1.7+, you must access __STACKTRACE__ inside a rescue/catch. If you want to support earlier Elixir versions, move System.stacktrace/0 inside a rescue/catch
  lib/opus/safe.ex:31

warning: System.stacktrace/0 outside of rescue/catch clauses is deprecated. If you want to support only Elixir v1.7+, you must access __STACKTRACE__ inside a rescue/catch. If you want to support earlier Elixir versions, move System.stacktrace/0 inside a rescue/catch
  lib/opus/safe.ex:37

warning: System.stacktrace/0 outside of rescue/catch clauses is deprecated. If you want to support only Elixir v1.7+, you must access __STACKTRACE__ inside a rescue/catch. If you want to support earlier Elixir versions, move System.stacktrace/0 inside a rescue/catch
  lib/opus/safe.ex:46

Generated opus app

I see CI uses 1.4/1.5. Do you plan to move minimal required version to at least 1.7 anytime?
Thanks

Environment

• opus version: 0.5.1

• Elixir / Hex version (mix hex.info):

  • Hex: 0.18.2
  • Elixir: 1.7.4
  • OTP: 21.1.1

• Operating system: MacOS 10.14.1

Current behavior

This is both a bug report and feature request. I think it's better to decouple Graphvix from Opus because it doesn't add anything for production. I'm also getting the below error on version 0.5 (version 1.0 is released)

[error] Graphvix.Graph Graphvix.Graph received unexpected message in handle_info/2: :save_state

Expected behavior

No error and decoupling

When trying to instrument the skip stage, we get a name that probably doesn't make too much sense (?).

Also, when the pipeline is skipped (skip true), Opus treat it as an error.

Environment

• opus version: 0.6.0

Current behavior

defmodule Test do
    use Opus.Pipeline
    
    skip if: :skip?

    def skip?(_), do: true

    def instrument(:stage_completed, params, params2) do
        require Logger
        Logger.warn("#{inspect params} >>> #{inspect params2}")
    end
end

Output:

%{stage: %{name: [if: :skip?], pipeline: Test}} >>> %{input: 1, result: {:error, :skipped}, stage: [if: :skip?], time: 31450000}

Expected behavior

1. Not sure what would be the expected behaviour in that case, since it's the only stage without a name?

2. Skip should not be an error

I find often times I'm just folding a map, and most of the code ends up just being getting a key out of the map and then passing it to some validation/execution function

Current behavior

defmodule MyOpus do
  use Opus.Pipeline
  
  check :valid_email?

  def valid_email?(%{email: email}) when is_binary(email) do
    MyValidator.valid_email?(email)
  end
end

Useful/Dry behavior

defmodule MyOpus do
  use Opus.Pipeline
  import MyValidator, only: [valid_email?: 1]
  
  # some accessor/getter/lens syntax
  check :valid_email?, in: :email
  # or
  check :valid_email?, get_in: [:email]
  # or
  check {:email, :valid_email?}
end

Is there a way for an if/unless step to modify the pipeline data?

I've found myself using pipelines similar to a with statement sometimes, and I would like some results to flow all the way to the end - even the conditional checks.
But I can't pass results of the conditional checks, as they only return a true/false, they don't mutate the pipeline.

An example might be to obtain the results for a pipeline that must complete successfully, it would never fully error out. I need to know that some steps were skipped, and the task/log result of why. I've seen instrumentation, but they are asynchronous. I need to feed the direct response and instantly reply with all conditional events skipped in the pipeline.

Any thoughts on this?

Proposal

Add the ability to pass a :unless parameter for all the stages where it is possible to pass a :if one.

The idea is that you don't need to name your functions like not_* in order to add a new stage to the pipeline.

Context

Imagine we can perform a step only if the user is allowed.

defmodule Pipeline do
  use Opus.Pipeline

  step :do_this, if: :user_allowed
end

Now, if we want to perform another step in case the user is not allowed, we would end up with something like:

defmodule Pipeline do
  use Opus.Pipeline

  step :do_this, if: :user_allowed
  step :do_another, if: :user_not_allowed
end

Which makes us implement two different functions (user_not_allowed and user_allowed) returning the opposite of the other. This would be simplified if we could do:

defmodule Pipeline do
  use Opus.Pipeline

  step :do_this, if: :user_allowed
  step :do_another, unless: :user_allowed
end

I want to pipe to the other opus pipeline but with behavior like tee ,
is there any function like that , or am I doing wrong ?
btw thx for this great library , it's like simplified saga pattern

Environment

• opus version: 0.5.3
• Elixir / Hex version (mix hex.info): 1.8
• Operating system: MacOS

Current behavior

When running my app I get the error above. My mix env:

{:opus, "~> 0.5.3"},
{:opus_graph, "~> 0.1", only: [:dev]},

Is it good practice to pipe a pipeline into another from within a stage? Some sort of conditional link.

My fix right now involves piping the results of one pipeline into another and checking a condition in the second stages first step and either returning pipeline or continuing with the steps.

I would like to use another module for handling instrumentation, the only issue is that it also implements instrument/3 and that ambiguity creates a problem, how can I use Opus without that function being in the scope I am in?

Current behavior

defmodule ArithmeticPipeline do
  use Opus.Pipeline

  step  :add_one,         with: &(&1 + 1)
  check :even?,           with: &(rem(&1, 2) == 0), error_message: :expected_an_even
  step :do_something_else

The error message should accept a function which will be called with the return value of the previous stage so that the error message can be parameterised.

Let's suppose we have a pipeline to create a user.

defmodule CreateUserPipeline do
  use Opus.Pipeline

  check :valid_params?
  step :fetch_avatar
  step :encrypt_password
  step :persist_user
  step :notify_clients

  # ...

Now, imagine that I don't want to run over all this pipeline if the user already exists. I can imagine two options to solve that:

1. Creating a call?/1 function in my module to be called before .call/1. This might work, but it forces the client of the module to know that. In my opinion, it can makes sense but sometimes not (sometimes, the pipeline itself should takes care of it as a defense).

2. Should I add a check at the beginning of the pipeline to halt the pipeline before all these steps? It can work, but then Opus will return a {:error, _} tuple. This might make sense most of the times, but sometimes it is not an error. It's just that all the params are correct but the pipeline should skip all the process of creating the user and return something like, {:ok, "User already exists"}.

So, in my opinion, it would be awesome to have a way to add a check at the beginning of the pipeline and add an option to return a custom tuple in case of false, like:

check :user_exists, on_false: {:ok, "User already exists"}

Makes sense?

I'm trying to reuse as much logic as possible in my pipelines.
As such, I'm having some problems when trying to work with them.

Let's imagine I have a message with several different kinds of data, for example: the same message has a device position (lat, lng), a state ("is the light turned on?") and a numeric value ("my current voltage is...")

Since the device sends all of those together, I'd need to parse and validate that message in a single pipeline.
And, since that data is already in the pipeline, I should be able to send each part of the message to a specific Pipeline to handle those.

For the position: StorePositionPipeline which stores the location and updates the "this is the current position of the device".

For the state: StoreStatePipeline which takes a state and a state name, and stores it.

For the numeric value: StoreNumericValuePipeline and VerifyVoltagePipeline. The first one is the same as StoreStatePipeline, but keeps track of numbers instead of enums. The second one has to do several things: check if the voltage is in the safe zone, if not maybe create an alert if there wasn't an alert already open for that device, ...

I am currently using a step and then calling the pipeline's call/1, since each one of those Pipelines expects its input in a different format.

By adding a transform to Link, I'd be able to write my pipeline as:

link StorePositionPipeline, transform: &extract_latlng/1
link StoreStatePipeline, transform: &extract_light_state/1
...

And that'd be much more readable IMHO than messing with steps just to transform the data and do a call.