Slither.Pipe.Runner (Slither v0.1.0)

Copy Markdown View Source

Executes a pipe definition against input data.

The runner:

  1. Creates or uses a session ID for Snakepit/SnakeBridge affinity
  2. Ensures store processes exist and registers views as session tools
  3. Executes stages sequentially, routing items between stages
  4. Collects results into output buckets
  5. Cleans up session tools on completion

Summary

Functions

run(pipe_mod, input, opts \\ [])

Execute a pipe against input, returning results grouped by output.

stream(pipe_mod, input, opts \\ [])

Execute a pipe returning a stream per output.

Functions

run(pipe_mod, input, opts \\ [])

@spec run(module(), Enumerable.t(), keyword()) ::
  {:ok, %{required(atom()) => [Slither.Item.t()]}}
  | {:error, term(), %{required(atom()) => [Slither.Item.t()]}}

Execute a pipe against input, returning results grouped by output.

Input can be raw payloads (auto-wrapped into Items) or pre-wrapped Items.

Options

  • :session_id - Session ID for Snakepit affinity (auto-generated if omitted)
  • :metadata - Additional metadata for the run context

stream(pipe_mod, input, opts \\ [])

@spec stream(module(), Enumerable.t(), keyword()) :: %{
  required(atom()) => Enumerable.t(Slither.Item.t())
}

Execute a pipe returning a stream per output.

Note: stages still execute eagerly per-batch within the stream, but the overall pipeline is lazy — batches flow through as demand allows.