sluice/sink

A Sink is the end of a pipeline. It subscribes to the stages before it and uses their events. Its speed controls the demand. Thus its speed controls the event flow through the full pipeline.

Types

Builder

opaque </>

pub opaque type Builder(state, event)

FoldError

</>

The result of a fold that did not complete.

pub type FoldError {
  FoldDidNotStart
  FoldDidNotSubscribe(reason: sluice.SubscribeError)
  FoldProducerFailed(reason: String)
  FoldTimeout
}

Constructors

```
FoldDidNotStart
```
The internal sink did not start.

FoldDidNotSubscribe(reason: sluice.SubscribeError)

The subscription to the outlet was not possible.

```
FoldProducerFailed(reason: String)
```
The producer failed before the end of the flow.
```
FoldTimeout
```
The producer did not stop in the given time.

Name

opaque </>

A permanent name for a sink. Make the name with new_name. Attach it with named. Find the sink that operates with inlet_of.

pub opaque type Name(event)

opaque </>

The response of a sink after it processes a batch of events. Make it with continue, stop, or stop_abnormal.

pub opaque type Next(state)

Values

continue

</>

pub fn continue(state: state) -> Next(state)

Continue with the new state.

state: The new state of the sink.

fold

</>

pub fn fold(
  from outlet: sluice.Outlet(event),
  initial initial: accumulated,
  with combine: fn(accumulated, event) -> accumulated,
  within timeout: Int,
) -> Result(accumulated, FoldError)

Run the full flow of an outlet through a fold, and return the final value when the producer stops. Use it together with from_yielder or an other source that has an end. The wait has a limit of within milliseconds.

outlet: The outlet that supplies the events.
initial: The first value of the accumulator.
combine: The function that folds one event into the accumulator.
timeout: The maximum wait time in milliseconds.

inlet_of

</>

pub fn inlet_of(name: Name(event)) -> sluice.Inlet(event)

The inlet of the sink that has this name. The inlet stays correct through restarts.

name: The name of the sink.

named

</>

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

Attach a permanent name to the sink 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(event), sluice.Subscription) -> Next(
    state,
  ),
) -> Builder(state, event)

Define a sink. The handler receives each batch of events and the subscription that supplied the batch.

state: The first state of the sink.
on_events: The batch handler. It receives the state, one batch of events, and the subscription that supplied the batch.

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.

on_cancelled

</>

pub fn on_cancelled(
  builder: Builder(state, event),
  on_cancelled: fn(state, sluice.SubscriptionEnd) -> Next(state),
) -> Builder(state, event)

Set a hook that runs when a subscription of the sink ends. When this hook is set, it decides what the sink does, and the cancel mode of the subscription does not apply.

builder: The builder to change.
on_cancelled: The hook. It receives the state and the SubscriptionEnd that gives the cause.

on_message

</>

pub fn on_message(
  builder: Builder(state, event),
  initialise initialise: fn(process.Subject(user_message)) -> Nil,
  handler handler: fn(state, user_message) -> Next(state),
) -> Builder(state, event)

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

on_subscribed

</>

pub fn on_subscribed(
  builder: Builder(state, event),
  on_subscribed: fn(state, sluice.Subscription) -> Next(state),
) -> Builder(state, event)

Set a hook that runs when the sink establishes a subscription. The hook receives the new Subscription, so a sink with manual demand can make its first ask here.

builder: The builder to change.
on_subscribed: The hook. It receives the state and the new Subscription.

start

</>

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

Start the sink. The returned data is its Inlet. You can subscribe it to outlets.

builder: The configuration of the sink.

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 sink, 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() -> Next(state)

Stop the sink with the normal reason. The stages before it see this through their monitors. They release the open demand of the sink.

stop_abnormal

</>

pub fn stop_abnormal(reason: String) -> Next(state)

Stop the sink with a failure reason.

reason: The description of the failure.

</>

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

Declare a subscription. The sink makes the connection during its start. The start fails for the checks that the sink can make itself: a dead producer, a duplicate, a subscription to itself, and demand values that are not correct. Under a supervisor, the restart sequence then does the retry. A refusal from the dispatcher of the producer, for example a missing partition, comes later as an abnormal end of the subscription. Use this together with outlet_of names for connections that continue through restarts.

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

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 sink.