consumer

Stateful OTP consumers with automatic subscription lifecycle handling.

Types

Consumer

opaque </>

A stateful consumer process subscribed to an OTP stage.

pub opaque type Consumer(state, event)

ConsumerError

</>

Failures from the consumer process or its Stage dependency.

pub type ConsumerError {
  Stage(otp.CallError)
  Timeout
  ConsumerUnavailable(process.ExitReason)
}

Constructors

```
Stage(otp.CallError)
```
```
Timeout
```
```
ConsumerUnavailable(process.ExitReason)
```

Values

ask

</>

pub fn ask(
  consumer: Consumer(state, event),
  amount: Int,
) -> Result(Nil, ConsumerError)

Adds demand for this consumer’s subscription.

start

</>

pub fn start(
  stage: otp.Stage(event),
  subscription_id: subscription_id.SubscriptionId,
  participant_id: participant_id.ParticipantId,
  initial_state: state,
  on_events: fn(state, List(event)) -> state,
) -> Result(
  actor.Started(Consumer(state, event)),
  actor.StartError,
)

Starts a consumer in the default partition.

start_partitioned

</>

pub fn start_partitioned(
  stage: otp.Stage(event),
  subscription_id: subscription_id.SubscriptionId,
  participant_id: participant_id.ParticipantId,
  partition: Int,
  initial_state: state,
  on_events: fn(state, List(event)) -> state,
) -> Result(
  actor.Started(Consumer(state, event)),
  actor.StartError,
)

Starts a consumer and subscribes it to a specific partition.

state

</>

pub fn state(
  consumer: Consumer(state, event),
) -> Result(state, ConsumerError)

Returns the consumer’s current application state.

stop

</>

pub fn stop(
  consumer: Consumer(state, event),
) -> Result(Nil, ConsumerError)

Cancels the subscription and stops the consumer process.