rx/runtime

Types

FlowKey

opaque </>

Identity for one asynchronous flattening flow owned by the Runtime actor.

pub opaque type FlowKey

FlowOrder

</>

Ordering policy for successful asynchronous completions.

pub type FlowOrder {
  CompletionOrder
  InputOrder
}

Constructors

```
CompletionOrder
```
Emit each result as soon as its asynchronous work completes.
```
InputOrder
```
Work may run concurrently, but buffer successful results until all earlier input sequence numbers have emitted.

Runtime

opaque </>

pub opaque type Runtime

RuntimeError

</>

pub type RuntimeError {
  RuntimeStopped
  InvalidConcurrency
}

Constructors

```
RuntimeStopped
```
```
InvalidConcurrency
```

SubscriptionKey

opaque </>

pub opaque type SubscriptionKey

Values

cancel

</>

pub fn cancel(runtime: Runtime, key: SubscriptionKey) -> Nil

cancel_flow

</>

pub fn cancel_flow(runtime: Runtime, key: FlowKey) -> Nil

complete_flow

</>

pub fn complete_flow(
  runtime: Runtime,
  key: FlowKey,
  sequence: Int,
  completion: FlowCompletion,
  deliver: fn() -> Nil,
) -> Nil

Report completion of an asynchronously running flow item.

deliver is a zero-argument typed closure capturing the successful value or error. This lets the runtime buffer/reorder work without erasing values to Dynamic.

dispatch

</>

pub fn dispatch(
  runtime: Runtime,
  key: SubscriptionKey,
  kind: protocol.Kind,
  work: fn() -> Nil,
) -> Nil

enqueue_flow

</>

pub fn enqueue_flow(
  runtime: Runtime,
  key: FlowKey,
  start: fn(Int) -> fn() -> Nil,
) -> Nil

Add one unit of work to a flow in FIFO input order.

The runtime assigns a monotonically increasing sequence number. start is invoked only when a concurrency slot is available and returns the physical cancellation callback for that unit of work.

fail_flow_input

</>

pub fn fail_flow_input(
  runtime: Runtime,
  key: FlowKey,
  deliver_error: fn() -> Nil,
) -> Nil

Fail the upstream input itself and cancel all active projected work.

finish_flow_input

</>

pub fn finish_flow_input(runtime: Runtime, key: FlowKey) -> Nil

register

</>

pub fn register(
  runtime: Runtime,
) -> Result(SubscriptionKey, RuntimeError)

Reserve one subscription entry without blocking on the runtime actor.

Registration is deliberately one-way so observer callbacks can subscribe to additional streams on the same runtime without self-deadlocking. The Register message is sent before producer code can emit or spawn work, so subsequent messages from that subscription are ordered behind registration.

register_flow

</>

pub fn register_flow(
  runtime: Runtime,
  concurrency: Int,
  order: FlowOrder,
  on_drain: fn() -> Nil,
) -> Result(FlowKey, RuntimeError)

This does not create another actor. Queue state is stored inside the existing Runtime actor. on_drain runs after the input has completed and all accepted work has completed and emitted.

Like subscription registration, flow registration is one-way and reentrant: callbacks running on the runtime actor may construct and subscribe nested flows without waiting for the actor to reply to itself.

set_teardown

</>

pub fn set_teardown(
  runtime: Runtime,
  key: SubscriptionKey,
  teardown: fn() -> Nil,
) -> Nil

start

</>

pub fn start() -> Result(Runtime, actor.StartError)

start_checked

</>

pub fn start_checked(
  on_diagnostic: fn(RuntimeDiagnostic) -> Nil,
) -> Result(Runtime, actor.StartError)

stop

</>

pub fn stop(runtime: Runtime) -> Nil

Stop the runtime after cancelling every active subscription and flow.

This call is intentionally asynchronous so it is safe to invoke from an observer callback running on the runtime actor itself. Active source teardowns and Future cancellation callbacks run before the actor exits.