Outboxer.Relay behaviour (Outboxer v0.1.0)

Copy Markdown View Source

The behaviour for a relay: the thing that takes messages out of the outbox and hands them to a broker, a queue or anything else.

Outboxer ships relays for Kafka (Outboxer.Relay.Kafka), SQS (Outboxer.Relay.SQS), Oban (Outboxer.Relay.Oban) and plain functions (Outboxer.Relay.Function). Each of the first three is compiled only when its dependency is in your project. Writing your own is a matter of implementing this behaviour; see the "Custom relays" guide.

Lifecycle

  1. init/1 runs once when the Outboxer supervisor starts. Validate the options and return the state that later calls receive. Raise or return {:error, reason} to refuse to start with a bad configuration.
  2. children/1 (optional) returns child specs to supervise next to the outbox workers, for example a broker client.
  3. deliver/2 is called with a batch of envelopes, in delivery order.

What deliver/2 returns

  • :ok - every message in the batch was delivered.
  • {:error, reason} - nothing was delivered.
  • {:partial, results} - a map of envelope id to :ok or {:error, reason}. Ids left out of the map count as "not attempted" and stay in the outbox untouched.

The ordering rule

For ordering to hold, a relay must not deliver a message after an earlier message of the same group has failed. Leave it out of the results (or report an error for it). sequential/2 implements this rule for relays that send one message at a time.

An exception raised from deliver/2 is treated as {:error, exception} for the whole batch.

Summary

Types

deliver_result()
result()
state()

Callbacks

children(state)

Child specs to supervise alongside the outbox workers.

deliver(envelopes, state)

Delivers a batch of envelopes, given in delivery order.

init(opts)

Validates the relay options and builds the relay state.

Functions

sequential(envelopes, fun)

Delivers envelopes one by one with fun, honouring the ordering rule.

Types

deliver_result()

@type deliver_result() ::
  :ok | {:error, term()} | {:partial, %{optional(pos_integer()) => result()}}

result()

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

state()

@type state() :: term()

Callbacks

children(state)

(optional) 
@callback children(state()) :: [Supervisor.child_spec() | {module(), term()} | module()]

Child specs to supervise alongside the outbox workers.

deliver(envelopes, state)

@callback deliver(envelopes :: [Outboxer.Envelope.t(), ...], state()) :: deliver_result()

Delivers a batch of envelopes, given in delivery order.

init(opts)

@callback init(opts :: keyword()) :: {:ok, state()} | {:error, term()}

Validates the relay options and builds the relay state.

Functions

sequential(envelopes, fun)

@spec sequential([Outboxer.Envelope.t()], (Outboxer.Envelope.t() -> result())) ::
  deliver_result()

Delivers envelopes one by one with fun, honouring the ordering rule.

fun receives an envelope and returns :ok or {:error, reason}. Once a message of a group fails, later messages of that group are skipped. Messages without a group never block anything.