sluice/source

A Source is the start of a pipeline. It makes events when the stages after it ask for them. Different processes can also push events into it through an Emitter.

Types

Builder

opaque </>

pub opaque type Builder(state, event)

Emitter

opaque </>

A handle for a source that operates. Each process can use it to push events into the source.

pub opaque type Emitter(event)

Name

opaque </>

A permanent name for a source. Subscribers can find the source through the name after a restart. Make the name with new_name. Attach it with named. Point subscriptions to it with outlet_of.

pub opaque type Name(event)

Produce

opaque </>

The response of a source to demand. Make it with emit, stop, or stop_abnormal.

pub opaque type Produce(state, event)

Values

accumulate_demand

</>

pub fn accumulate_demand(
  builder: Builder(state, event),
) -> Builder(state, event)

Start the source in the accumulation mode: incoming asks wait, and the source makes no events. A call of sluice.forward_demand on the outlet releases the held asks. Use this to connect a full pipeline before the events start to move.

builder: The builder to change.

buffer_capacity

</>

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

Set the maximum quantity of events that the buffer keeps while there is no demand. The default is 10,000.

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

buffer_keep

</>

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

Select the events that stay when the buffer is full. The default is KeepLast.

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

buffer_unbounded

</>

pub fn buffer_unbounded(
  builder: Builder(state, event),
) -> Builder(state, event)

Remove the buffer limit.

builder: The builder to change.

dispatcher

</>

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

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

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

emit

</>

pub fn emit(
  events events: List(event),
  state state: state,
) -> Produce(state, event)

Emit events for the demand. The quantity of events can be more than the demand: the buffer keeps the extra events. The quantity can also be less than the demand, or zero. The demand that stays open is then supplied by later events, from a push through an Emitter or from a message handler that emits.

Emit less than the demand only when such later events can come: without them, the consumers wait for the open demand and do not ask again. A source that ends must end with emit_final.

events: The events to emit.
state: The new state of the source.

emit_final

</>

pub fn emit_final(events: List(event)) -> Produce(state, event)

Emit the last events, and then stop the source with the normal reason. Use this when the source ends in the middle of a demand: the events go out first, and then the subscribers apply their cancel modes.

events: The last events of the source.

finish

</>

pub fn finish(emitter: Emitter(event)) -> Nil

Stop the source with the normal reason, from a different process.

emitter: The Emitter of the source.

from_yielder

</>

pub fn from_yielder(
  yielder yielder: yielder.Yielder(event),
) -> Builder(yielder.Yielder(event), event)

Define a source that takes its events from a yielder. The source steps the yielder as far as the demand asks. It stops with the normal reason at the end of the yielder.

yielder: The yielder that supplies the events.

named

</>

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

Attach a permanent name to the source at its start.

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

new

</>

pub fn new(
  init state: state,
  on_demand on_demand: fn(state, Int) -> Produce(state, event),
) -> Builder(state, event)

Define a source that makes events on demand.

state: The first state of the source.
on_demand: The demand handler. It receives the state and the open demand, and it returns a Produce with the new events.

new_name

</>

pub fn new_name(prefix prefix: String) -> Name(event)

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

prefix: The readable prefix of the name.

new_with_emitter

</>

pub fn new_with_emitter(
  init initialise: fn(Emitter(event)) -> Result(state, String),
) -> Builder(state, event)

Define a source that receives its events from an external location. The initialiser receives an Emitter. Each process can push events through the Emitter. A source that only receives pushes can keep the default on_demand. If the source can also make events on request, set a handler with on_demand.

initialise: The initialiser. It receives the Emitter of the source, and it returns the first state.

on_demand

</>

pub fn on_demand(
  builder: Builder(state, event),
  on_demand: fn(state, Int) -> Produce(state, event),
) -> Builder(state, event)

Set the demand handler of a source that new_with_emitter made.

builder: The builder to change.
on_demand: The demand handler. It receives the state and the open demand, and it returns a Produce with the new events.

on_discard

</>

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

Set the response of the source 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, event),
  initialise initialise: fn(process.Subject(user_message)) -> Nil,
  handler handler: fn(state, user_message) -> Produce(
    state,
    event,
  ),
) -> Builder(state, event)

Give the source 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, like the demand handler. 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 source.
handler: The message handler. It receives the state and one message, and it returns a Produce.

on_subscribers

</>

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

Set a hook that runs when a subscriber arrives or leaves. Use it, for example, to start work at the first subscriber and to stop work at the last one.

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

outlet_of

</>

pub fn outlet_of(name: Name(event)) -> sluice.Outlet(event)

The outlet of the source that has this name. The outlet stays correct through restarts of the source. Thus use it to connect stages under a supervisor.

name: The name of the source.

push

</>

pub fn push(
  through emitter: Emitter(event),
  events events: List(event),
) -> Nil

Push events into the source. The source sends them to the consumers immediately, up to the open demand. The buffer keeps the remaining events.

emitter: The Emitter of the source.
events: The events to push.

start

</>

pub fn start(
  builder: Builder(state, event),
) -> Result(actor.Started(sluice.Outlet(event)), actor.StartError)

Start the source. The returned data is its Outlet. You can subscribe to it.

builder: The configuration of the source.

start_timeout

</>

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

The maximum time for the start of the source. The default is 5000 milliseconds.

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

stop

</>

pub fn stop() -> Produce(state, event)

Stop the source with the normal reason. The subscribers apply their cancel modes.

The demand handler runs again only when a consumer asks again, and a consumer asks again only after it received its full demand. Thus stop can end the source only after emits that filled the demand completely. When the events end in the middle of a demand, use emit_final in that same call.

stop_abnormal

</>

pub fn stop_abnormal(reason: String) -> Produce(state, event)

Stop the source with a failure reason.

reason: The description of the failure.

whereis

</>

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

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

name: The name of the source.