distribute/receiver

Types

Start an actor that receives and decodes typed messages.

This is a convenience wrapper around actor.start that handles binary decoding automatically. The returned Subject accepts BitArray messages, making it suitable for use with distribute’s typed messaging API.

Malformed messages are silently ignored.

Note: This helper does not support custom selectors in the handler return value (they will be ignored). If you need complex selector logic, use selecting_typed with a standard actor.start_spec.

Parameters

initial_state: The initial state of the actor.
decoder: The decoder for incoming messages.
handler: The message handler function.

Returns

Ok(Subject(BitArray)) on success, or Error(StartError). The return value for the handler function in start_typed_receiver.

pub type Next(state) {
  Continue(state: state)
  Stop
  StopAbnormal(reason: String)
}

Constructors

```
Continue(state: state)
```
Continue handling messages with the new state.
```
Stop
```
Stop the actor normally.
```
StopAbnormal(reason: String)
```
Stop the actor abnormally with a reason.

ReceiveError

</>

Errors that can occur when receiving messages.

pub type ReceiveError {
  DecodeError(codec.DecodeError)
  Timeout
}

Constructors

```
DecodeError(codec.DecodeError)
```
The message could not be decoded.
```
Timeout
```
No message was received before the timeout.

Values

receive_loop_typed

</>

pub fn receive_loop_typed(
  subject: process.Subject(BitArray),
  decoder: fn(BitArray) -> Result(a, codec.DecodeError),
  initial_state: state,
  handler: fn(a, state) -> state,
) -> Nil

Receive messages in a loop with a typed handler function.

This function continuously receives messages from a Subject, decodes them, and passes them to a handler function. The handler returns updated state, and the loop continues until externally stopped (process termination).

Malformed messages that fail to decode are silently skipped.

Parameters

subject: The Subject to receive messages from.
decoder: The Decoder for incoming binary messages.
initial_state: The initial state value.
handler: Function that processes decoded messages and returns updated state.

Note

This function runs forever. Use it in a spawned process that can be terminated externally, or structure your handler to perform side effects and ignore state if you want graceful shutdown.

receive_typed

</>

pub fn receive_typed(
  subject: process.Subject(BitArray),
  decoder: fn(BitArray) -> Result(a, codec.DecodeError),
  timeout_ms: Int,
) -> Result(a, ReceiveError)

Receive a single typed message from a Subject with timeout.

Waits for a message, decodes it using the provided decoder, and returns the decoded value. Returns an error on timeout or decode failure.

selecting_typed

</>

pub fn selecting_typed(
  selector: process.Selector(b),
  subject: process.Subject(BitArray),
  decoder: fn(BitArray) -> Result(a, codec.DecodeError),
  mapper: fn(Result(a, ReceiveError)) -> b,
) -> process.Selector(b)

Add a typed message handler to a Selector.

This function allows you to add a handler for messages from a specific Subject to an existing Selector. The handler will decode messages using the provided decoder and transform them with the mapper function.

The mapper function receives a Result containing either the decoded value or a decode error. This allows the caller to handle malformed messages gracefully (e.g. by logging them or ignoring them).

Parameters

selector: The Selector to add the handler to.
subject: The Subject to receive messages from.
decoder: The Decoder to use for decoding binary messages.
mapper: A function to transform the decode result into the selector’s return type b.

Returns

The updated Selector with the new handler added.

start_global_receiver

</>

pub fn start_global_receiver(
  initial_state: state,
  decoder: fn(BitArray) -> Result(msg, codec.DecodeError),
  handler: fn(msg, state) -> Next(state),
) -> process.Subject(BitArray)

Start a global receiver actor.

This actor is compatible with registry.register_typed and registry.whereis_typed. It accepts messages with a Nil tag, which allows it to be addressed globally without sharing a specific Reference tag.

Use this function if you intend to register the process using registry.register_typed.

start_typed_actor

</>

pub fn start_typed_actor(
  initial_state: state,
  encoder: fn(msg) -> Result(BitArray, codec.EncodeError),
  decoder: fn(BitArray) -> Result(msg, codec.DecodeError),
  handler: fn(msg, state) -> Next(state),
) -> global.GlobalSubject(msg)

Start a typed actor that returns a GlobalSubject (RECOMMENDED).

This is the recommended way to create type-safe actors for distributed use. The returned GlobalSubject enforces encoder/decoder usage and can be registered globally with registry.register_typed.

Malformed messages are silently ignored.

Parameters

initial_state: The initial state of the actor.
encoder: The encoder for outgoing messages (used by clients).
decoder: The decoder for incoming messages.
handler: The message handler function returning Next(state).

Returns

A GlobalSubject(msg) that can be used for type-safe messaging.

Example

pub type Request {
  GetCount
  Increment
}

let actor = receiver.start_typed_actor(
  0,
  my_encoder(),
  my_decoder(),
  fn(msg, count) {
    case msg {
      GetCount -> {
        // Send response logic here
        receiver.Continue(count)
      }
      Increment -> receiver.Continue(count + 1)
    }
  },
)

start_typed_receiver

</>

pub fn start_typed_receiver(
  initial_state: state,
  decoder: fn(BitArray) -> Result(msg, codec.DecodeError),
  handler: fn(msg, state) -> Next(state),
) -> Result(process.Subject(BitArray), actor.StartError)

Start an actor that receives and decodes typed messages.

Malformed messages are silently ignored.

Note: This helper does not support custom selectors in the handler return value. If you need complex selector logic, use selecting_typed with a standard actor.start_spec.

Parameters

initial_state: The initial state of the actor.
decoder: The decoder for incoming messages.
handler: The message handler function.

Returns

Ok(Subject(BitArray)) on success, or Error(StartError).