sluice/gate

A Gate is in the middle of a pipeline. It uses events from the stages before it. It changes the events. It makes the results for the stages after it. Its two faces are connected: the gate asks for more events only while the stages after it ask for events. Thus backpressure goes through the gate.

Types

Builder

opaque </>

pub opaque type Builder(state, in, out)

Gate

opaque </>

The two faces of a gate that operates. Subscribe its inlet to the outlets before it. Subscribe the subsequent inlets to its outlet.

pub opaque type Gate(in, out)

Name

opaque </>

A permanent name for a gate. It gives the two faces addresses that stay correct through restarts: inlet_of and outlet_of.

pub opaque type Name(in, out)

Transform

opaque </>

The response of a gate to a batch of events. Make it with emit, stop, or stop_abnormal.

pub opaque type Transform(state, out)

Values

buffer_capacity

</>

pub fn buffer_capacity(
  builder builder: Builder(state, in, out),
  events capacity: Int,
) -> Builder(state, in, out)

Set a limit for the output buffer of the gate. The default is no limit. The demand connection between the two faces keeps the buffer small. The maximum content is approximately the sum of the max_demand values of the subscriptions before the gate. Thus the gate discards nothing.

builder: The builder to change.
capacity: The maximum quantity of events in the buffer.

buffer_keep

</>

pub fn buffer_keep(
  builder builder: Builder(state, in, out),
  keep keep: sluice.Keep,
) -> Builder(state, in, out)

Select the events that stay when a buffer with a limit is full.

builder: The builder to change.
keep: The Keep selection.

dispatcher

</>

pub fn dispatcher(
  builder builder: Builder(state, in, out),
  dispatcher dispatcher: dispatcher.Dispatcher(out),
) -> Builder(state, in, out)

Set the dispatcher of the gate. The default is the demand dispatcher.

builder: The builder to change.
dispatcher: The dispatcher of the gate.

emit

</>

pub fn emit(
  events events: List(out),
  state state: state,
) -> Transform(state, out)

Emit the changed events. The quantity of output events is free. A filter makes fewer events. An expander makes more events. The buffer keeps the events that are more than the demand after the gate.

events: The changed events.
state: The new state of the gate.

inlet

</>

pub fn inlet(gate: Gate(in, out)) -> sluice.Inlet(in)

The input face of the gate.

gate: The gate, from start.

inlet_of

</>

pub fn inlet_of(name: Name(in, out)) -> sluice.Inlet(in)

The input face of the gate that has this name. It stays correct through restarts.

name: The name of the gate.

named

</>

pub fn named(
  builder builder: Builder(state, in, out),
  name name: Name(in, out),
) -> Builder(state, in, out)

Attach a permanent name to the gate at its start.

builder: The builder to change.
name: The name, from new_name.

new

</>

pub fn new(
  init state: state,
  on_events on_events: fn(state, List(in), sluice.Subscription) -> Transform(
    state,
    out,
  ),
) -> Builder(state, in, out)

Define a gate. The handler receives each batch and the subscription that supplied the batch. The handler returns the changed events.

state: The first state of the gate.
on_events: The batch handler. It receives the state, one batch of events, and the subscription that supplied the batch, and it returns a Transform with the changed events.

new_name

</>

pub fn new_name(prefix prefix: String) -> Name(in, out)

Make a permanent name. Create names during application start, not inside a dynamic loop.

prefix: The readable prefix of the name.

on_discard

</>

pub fn on_discard(
  builder: Builder(state, in, out),
  on_discard: fn(state, Int) -> state,
) -> Builder(state, in, out)

Set the response of the gate to discarded events. The callback receives the state and the quantity of discarded events. The default response writes a warning to the log.

builder: The builder to change.
on_discard: The callback. It receives the state and the quantity of discarded events, and it returns the new state.

on_message

</>

pub fn on_message(
  builder: Builder(state, in, out),
  initialise initialise: fn(process.Subject(user_message)) -> Nil,
  handler handler: fn(state, user_message) -> Transform(
    state,
    out,
  ),
) -> Builder(state, in, out)

Give the gate a private message channel with a type of your choice. At the start, initialise receives the subject of the channel: send it to other processes, or start a timer with process.send_after. The handler receives each message together with the state, and it can emit events to the stages after the gate. Use this for timers, for configuration changes, and for queries.

builder: The builder to change.
initialise: The function that receives the subject of the channel at the start of the gate.
handler: The message handler. It receives the state and one message, and it returns a Transform.

on_subscribers

</>

pub fn on_subscribers(
  builder: Builder(state, in, out),
  on_subscribers: fn(state, sluice.SubscriberChange) -> state,
) -> Builder(state, in, out)

Set a hook that runs when a subscriber arrives at the gate or leaves it.

builder: The builder to change.
on_subscribers: The hook. It receives the state and the SubscriberChange, and it returns the new state.

outlet

</>

pub fn outlet(gate: Gate(in, out)) -> sluice.Outlet(out)

The output face of the gate.

gate: The gate, from start.

outlet_of

</>

pub fn outlet_of(name: Name(in, out)) -> sluice.Outlet(out)

The output face of the gate that has this name. It stays correct through restarts.

name: The name of the gate.

start

</>

pub fn start(
  builder: Builder(state, in, out),
) -> Result(actor.Started(Gate(in, out)), actor.StartError)

Start the gate. The returned data is the pair of faces. Connect them with inlet and outlet.

builder: The configuration of the gate.

start_timeout

</>

pub fn start_timeout(
  builder builder: Builder(state, in, out),
  milliseconds milliseconds: Int,
) -> Builder(state, in, out)

The maximum time for the start of the gate, which includes its declared subscriptions. The default is 5000 milliseconds.

builder: The builder to change.
milliseconds: The maximum start time in milliseconds.

stop

</>

pub fn stop() -> Transform(state, out)

Stop the gate with the normal reason.

stop_abnormal

</>

pub fn stop_abnormal(reason: String) -> Transform(state, out)

Stop the gate with a failure reason.

reason: The description of the failure.

</>

pub fn subscribe(
  builder builder: Builder(state, in, out),
  options options: sluice.SubscriptionOptions(in),
) -> Builder(state, in, out)

Declare a subscription. The gate makes the connection during its start. The start fails for the checks that the gate can make itself: a dead producer, a duplicate, a subscription to itself, and demand values that are not correct. A refusal from the dispatcher of the producer comes later as an abnormal end of the subscription.

builder: The builder to change.
options: The subscription options, from sluice.subscription.

whereis

</>

pub fn whereis(name: Name(in, out)) -> Result(process.Pid, Nil)

The process that has this name now, if a process has it.

name: The name of the gate.