class

Arcana::Mailbox

A buffered inbox for a single agent address.

Backed by a Deque (not a Channel) so messages can be inspected without consuming them and selectively received by id.

Optionally bounded: pass max_queue: to reject deliveries with MailboxFull when the queue reaches capacity. Unbounded by default — the daemon sets a global default via ARCANA_MAILBOX_MAX_QUEUE.

Constructors

new(address : String, max_queue : Int32 | Nil = nil)

Source

Instance methods

address

Source

await_outstanding(timeout : Time::Span) : Bool

Block until all expectations are met or timeout expires. Returns true if all met, false on timeout.

Source

deliver(envelope : Envelope)

Deliver an envelope to this mailbox. Raises MailboxFull when max_queue is set and the queue is at capacity.

Source

dump

Snapshot the mailbox's full state for persistence.

Source

events

Optional event recorder. Bus sets this on newly created mailboxes so deliver/consume/freeze/thaw emit audit events.

Source

events=(events : Events::Backend | Nil)

Optional event recorder. Bus sets this on newly created mailboxes so deliver/consume/freeze/thaw emit audit events.

Source

expect(correlation_id : String)

Source

freeze(id : String, by : String = "") : Bool

Freeze a message: move from deque to frozen map. Returns true if a message was frozen.

Source

frozen

List frozen message metadata (non-destructive).

Source

frozen_count

Count of frozen messages.

Source

fulfill(correlation_id : String) : Bool

Mark an expectation as fulfilled. Called automatically by deliver.

Source

inbox

Non-destructive listing of message metadata.

Source

last_activity

Last time this mailbox had a deliver or receive operation.

Source

last_activity=(time : Time)

Set last activity timestamp directly (used by snapshot restore).

Source

load_snapshot(messages : Array(Envelope), frozen : Hash(String, Envelope), frozen_by : Hash(String, String))

Restore mailbox state from a snapshot. Replaces existing state.

Source

max_queue

Source

on_activity

Optional callback invoked on any activity (deliver, receive). Used by Bus to refresh Directory last_seen.

Source

on_activity=(on_activity : Proc(String, Nil) | Nil)

Optional callback invoked on any activity (deliver, receive). Used by Bus to refresh Directory last_seen.

Source

outstanding

Count of unfulfilled expectations.

Source

pending

Number of messages waiting to be received.

Source

receive(id : String, timeout : Time::Span) : Envelope | Nil

Block until a specific message (by correlation_id) arrives or timeout expires. Returns nil on timeout.

Source

receive(timeout : Time::Span) : Envelope | Nil

Block until an envelope arrives or timeout expires. Returns nil on timeout.

Source

receive(id : String) : Envelope | Nil

Receive a specific message by correlation_id. Returns nil if not found.

Source

receive

Block until an envelope arrives.

Source

thaw(id : String) : Envelope | Nil

Thaw a frozen message: move back to deque. Returns the thawed envelope, or nil if not found.

Source

thaw_all

Thaw all frozen messages back to the deque.

Source

try_receive

Non-blocking receive. Returns nil if empty.

Source