Sig.Node behaviour (Sig v0.1.0)

Copy Markdown View Source

Behaviour for workflow nodes.

A node is a module that implements either run/1 or run/2. It receives the merged params, context and the results of the nodes it depends on, and returns its own result.

run/1

Receives the whole Sig.NodeContext:

defmodule MyApp.Nodes.CalculateSum do
  use Sig.Node

  @impl true
  def run(context) do
    a = context.params["a"] || 0
    b = context.params["b"] || 0
    {:ok, %{"sum" => a + b}}
  end
end

run/2

Receives the merged params as the first argument and the Sig.NodeContext as the second:

defmodule MyApp.Nodes.FormatResult do
  use Sig.Node

  @impl true
  def run(params, context) do
    sum = context.results["calculate_sum"]["sum"]
    prefix = params["prefix"] || "Total:"
    {:ok, %{"formatted" => "#{prefix} #{sum}"}}
  end
end

When a module defines both, run/1 is used.

Return values

  • {:ok, result} - the node succeeded. result is stored and exposed to downstream nodes as context.results[node_name].
  • {:error, reason} - the node failed. It is retried according to its max_retries/backoff_ms; when no attempts are left, the execution is marked as failed with reason as its error.
  • any other value - treated as {:ok, value}.

Raised exceptions, throws and exits are caught and handled like {:error, reason}, with the exception message and stacktrace stored as the error.

Results are persisted as JSON: atom keys and values become strings, tuples become lists, and non-map results are wrapped as %{"value" => result}. See Sig.Serializer.

Idempotency

A node that completed is never run again for the same execution, even when the execution is resumed. A node that fails mid-way, however, may be retried, so side effects should be idempotent (use context.execution_id and context.node_name as an idempotency key for external calls).

Summary

Types

result()

The value returned by a node.

Callbacks

run(context)

Runs the node with its full Sig.NodeContext.

run(params, context)

Runs the node with its merged params and its Sig.NodeContext.

Functions

__using__(opts)

Declares the current module as a Sig.Node.

execute(module, context)

Runs a node module with the given Sig.NodeContext.

resolve_module(module_string)

Resolves a module name to a loaded module.

Types

result()

@type result() :: {:ok, term()} | {:error, term()} | term()

The value returned by a node.

Callbacks

run(context)

(optional) 
@callback run(context :: Sig.NodeContext.t()) :: result()

Runs the node with its full Sig.NodeContext.

run(params, context)

(optional) 
@callback run(params :: map(), context :: Sig.NodeContext.t()) :: result()

Runs the node with its merged params and its Sig.NodeContext.

params is the same as context.params.

Functions

__using__(opts)

(macro)

Declares the current module as a Sig.Node.

use Sig.Node

execute(module, context)

@spec execute(module() | String.t(), Sig.NodeContext.t()) ::
  {:ok, term()} | {:error, term()}

Runs a node module with the given Sig.NodeContext.

module may be a module or its name as a string (as stored in the workflow). Dispatches to run/1 or run/2, normalizes the return value to {:ok, result} or {:error, reason} and turns exceptions, throws and exits into errors.

This is used by Sig.Engine; you normally do not call it directly, but it is handy for unit-testing a node in isolation:

context = Sig.NodeContext.build(%{node_name: "sum", node_params: %{"a" => 1, "b" => 2}})
{:ok, %{"sum" => 3}} = Sig.Node.execute(MyApp.Nodes.CalculateSum, context)

resolve_module(module_string)

@spec resolve_module(String.t() | module()) :: {:ok, module()} | {:error, String.t()}

Resolves a module name to a loaded module.

Accepts names with or without the Elixir. prefix. Only existing atoms are used, so arbitrary strings never create new atoms.

Examples

iex> Sig.Node.resolve_module("Enum")
{:ok, Enum}

iex> Sig.Node.resolve_module("Does.Not.Exist")
{:error, "Module Does.Not.Exist does not exist"}