constellation - v0.1.0

constellation/runtime/otp

Typed OTP process adapter for Constellation stages.

Types

CallError

</>

Failures from the OTP transport or the pure stage runtime.

A timed-out operation may still complete because its message may already be in the Stage mailbox. Callers must not assume that Timeout means the operation was not applied.

pub type CallError {
  Runtime(runtime.RuntimeError)
  Timeout
  StageUnavailable(process.ExitReason)
}

Constructors

Config

opaque </>

Configuration for an OTP stage.

pub opaque type Config(event)

ConfigError

</>

Invalid OTP Stage configuration values.

pub type ConfigError {
  InvalidCallTimeout(Int)
  InvalidBufferCapacity(Int)
}

Constructors 

  • InvalidCallTimeout(Int)
  • InvalidBufferCapacity(Int)

Stage

opaque </>

A running OTP stage. Its actor subject is intentionally hidden.

pub opaque type Stage(event)

Values

ask

</> 
pub fn ask(
  stage: Stage(event),
  subscription_id: subscription_id.SubscriptionId,
  amount: Int,
) -> Result(Nil, CallError)

Adds downstream demand and synchronously returns validation errors.

cancel

</> 
pub fn cancel(
  stage: Stage(event),
  subscription_id: subscription_id.SubscriptionId,
) -> Result(Nil, CallError)

Cancels a subscription and delivers its cancellation notification.

config

</> 
pub fn config() -> Config(event)

Returns the default demand-dispatch stage configuration.

push

</> 
pub fn push(
  stage: Stage(event),
  events: List(event),
) -> Result(Nil, CallError)

Pushes events and executes any resulting outbound deliveries.

report_consumed

</> 
pub fn report_consumed(
  stage: Stage(event),
  subscription_id: subscription_id.SubscriptionId,
  amount: Int,
) -> Nil

Reports that a consumer finished processing a previously received batch.

This is an observability signal, not a protocol acknowledgement. Consumers call it only after their own message handler has completed its work.

start

</> 
pub fn start() -> Result(
  actor.Started(Stage(event)),
  actor.StartError,
)

Starts a stage using demand-based round-robin dispatching.

start_with_config

</> 
pub fn start_with_config(
  config: Config(event),
) -> Result(actor.Started(Stage(event)), actor.StartError)

Starts a stage actor with explicit configuration.

start_with_strategy

</> 
pub fn start_with_strategy(
  strategy: dispatcher.Strategy(event),
) -> Result(actor.Started(Stage(event)), actor.StartError)

Starts a stage actor with the supplied pure dispatch strategy.

stop

</> 
pub fn stop(stage: Stage(event)) -> Result(Nil, CallError)

Stops the stage actor after all earlier messages have been handled.

subscribe

</> 
pub fn subscribe(
  stage: Stage(event),
  id: subscription_id.SubscriptionId,
  participant_id: participant_id.ParticipantId,
  partition: Int,
  recipient: process.Subject(runtime.ParticipantMessage(event)),
) -> Result(Nil, CallError)

Registers a participant subject and creates its core subscription.

with_buffer_capacity

</> 
pub fn with_buffer_capacity(
  config: Config(event),
  capacity: Int,
) -> Result(Config(event), ConfigError)

Sets the maximum number of events that may wait in the Stage buffer.

with_call_timeout

</> 
pub fn with_call_timeout(
  config: Config(event),
  milliseconds: Int,
) -> Result(Config(event), ConfigError)

Sets the maximum time for synchronous OTP calls.

This does not change the timeout used to start the Stage actor.

with_logging

</> 
pub fn with_logging(config: Config(event)) -> Config(event)

Enables lifecycle logging through the standard Erlang logger.

with_reporter

</> 
pub fn with_reporter(
  config: Config(event),
  reporter: fn(telemetry.Event) -> Nil,
) -> Config(event)

Sends structured lifecycle events to a custom reporter.

The reporter runs inside the stage process and must return quickly.

with_strategy

</> 
pub fn with_strategy(
  config: Config(event),
  strategy: dispatcher.Strategy(event),
) -> Config(event)

Replaces the dispatch strategy in a stage configuration.

✨ Search Document