Sourced.Middleware behaviour (sourced v0.1.0)

Copy Markdown View Source

A composable transformation wrapped around an event store's adapter call.

Middleware is how Sourced.EventStore stays agnostic of concerns like domain event conversion and telemetry. Each middleware receives the Sourced.Operation flowing through the pipeline, a next value standing for the rest of the stack, and its own configured opts. It may transform the operation on the way in, must call next/2 to continue, and may transform the returned operation's result on the way out:

defmodule MyApp.Tracing do
  @behaviour Sourced.Middleware

  @impl true
  def init(opts), do: opts

  @impl true
  def call(op, opts, next) do
    op = %{op | opts: annotate(op.opts, opts)}
    Sourced.Middleware.next(next, op)
  end
end
end

Configuring a pipeline

Entries passed to Sourced.EventStore.new/1 via :middleware are either a bare module or a {module, opts} tuple (Plug-style). The first entry is the outermost wrapper; the adapter call is the innermost terminal:

Sourced.EventStore.new(
  adapter: Sourced.EventStore.Postgres,
  middleware: [
    {Sourced.Middleware.Domain, [OrderPlaced, OrderShipped]},
    {MyApp.Tracing, sample: 0.1}
  ]
)

Initialization

Each entry's opts are passed through the middleware's init/1 when the store is built, and it is init/1's return value that call/3 receives on every operation. This is where option validation and any expensive preprocessing belong, so the work happens once per store rather than on every operation — see Sourced.Middleware.Domain for an example. Implementing init/1 is optional; without it the opts are passed through untouched.

Summary

Types

A pipeline entry: a bare module (empty opts) or a {module, opts} tuple.

An entry whose options have been through init/1.

The remainder of the pipeline, to be handed back to next/2.

Initialized entries, outermost first.

Callbacks

Transforms op, invokes next/2, and returns the resulting operation.

Prepares the entry's configured options, when the store is built.

Functions

Called by Sourced.EventStore.new/1 to initialize a list of entries.

Invokes the remainder of the pipeline, terminating at the adapter call.

Runs an event store operation through the middleware pipeline.

Types

entry()

@type entry() :: module() | {module(), term()}

A pipeline entry: a bare module (empty opts) or a {module, opts} tuple.

initialized()

@type initialized() :: {module(), term()}

An entry whose options have been through init/1.

next()

@type next() :: pipeline()

The remainder of the pipeline, to be handed back to next/2.

pipeline()

@type pipeline() :: [initialized()]

Initialized entries, outermost first.

Callbacks

call(op, opts, next)

@callback call(op :: Sourced.Operation.t(), opts :: term(), next :: next()) ::
  Sourced.Operation.t()

Transforms op, invokes next/2, and returns the resulting operation.

opts is the middleware's initialized options (see init/1). The operation returned by Sourced.Middleware.next(next, op) is the one carrying result.

init(opts)

(optional)
@callback init(opts :: term()) :: term()

Prepares the entry's configured options, when the store is built.

Whatever it returns is what call/3 receives as its opts on every operation. Optional: entries whose module does not implement it keep their options as given ([] for a bare-module entry).

Functions

build_pipeline(entries)

@spec build_pipeline([entry()]) :: pipeline()

Called by Sourced.EventStore.new/1 to initialize a list of entries.

The result is plain data, which is what lets a store be held in a module attribute: the layers are walked by next/2 on each operation rather than composed into a closure that a module attribute could not hold.

next(list, operation)

@spec next(next :: next(), operation :: Sourced.Operation.t()) ::
  Sourced.Operation.t()

Invokes the remainder of the pipeline, terminating at the adapter call.

Every call/3 implementation must call this with the next it was handed, and return the operation it produces.

run(operation, pipeline)

@spec run(operation :: Sourced.Operation.t(), pipeline :: pipeline()) ::
  Sourced.Operation.t()

Runs an event store operation through the middleware pipeline.

A :notify operation carries events backwards from the adapter.