class

Arcana::Bus

Inherits Reference < Object

Central message router for agent-to-agent communication.

Supports direct delivery (send) and fan-out (publish/subscribe). Addresses are resolved through the directory when available — bare names like "memo" resolve to "memo:agent" or "memo:service" if unambiguous.

bus = Arcana::Bus.new writer = bus.mailbox("writer:agent") artist = bus.mailbox("artist:agent")

Direct message — bare name resolves if unambiguous

bus.send(Envelope.new(from: "writer:agent", to: "artist", ...))

Instance methods

addresses

List all registered addresses.

Source
default_max_queue

Default max queue length for newly created mailboxes. nil means unbounded (the historical default). Set via bin/arcana.cr from ARCANA_MAILBOX_MAX_QUEUE. Mailboxes created with an explicit max_queue: via a custom mailbox_factory override this.

Source
default_max_queue=(default_max_queue : Int32 | Nil)

Default max queue length for newly created mailboxes. nil means unbounded (the historical default). Set via bin/arcana.cr from ARCANA_MAILBOX_MAX_QUEUE. Mailboxes created with an explicit max_queue: via a custom mailbox_factory override this.

Source
deliver(envelope : Envelope, timeout : Time::Span = 30.seconds) : Tuple(Envelope | Nil, Ordering)

Dispatch based on the envelope's ordering field (auto-resolved). Returns {reply, resolved_ordering}. Reply is nil for async.

Source
deliver?(envelope : Envelope, timeout : Time::Span = 30.seconds) : Tuple(Envelope | Nil, Ordering)

Like deliver, but silently drops if the target mailbox doesn't exist.

Source
directory
Source
directory=(directory : Directory | Nil)
Source
events

Optional event recorder. When set, material bus actions (sends, publishes, subscribe/unsubscribe, prune) emit events. Newly created mailboxes inherit this recorder via their persistence hooks.

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

Optional event recorder. When set, material bus actions (sends, publishes, subscribe/unsubscribe, prune) emit events. Newly created mailboxes inherit this recorder via their persistence hooks.

Source
has_mailbox?(address : String) : Bool

Does a mailbox exist for this address?

Source
mailbox(address : String) : Mailbox

Get or create a mailbox for an address.

Source
mailbox_factory
Source
mailbox_factory=(mailbox_factory : MailboxFactory)
Source
pending(address : String) : Int32

Pending message count for an address. Returns 0 if no mailbox.

Source
prune_stale(listing_ttl : Time::Span, mailbox_ttl : Time::Span) : Tuple(Array(String), Array(String))

Prune stale agent listings and inactive mailboxes.

  • Agent listings with last_seen older than listing_ttl are removed.
  • Mailboxes with last_activity older than mailbox_ttl are removed. Services are never pruned (they are re-registered from code at startup). Returns {pruned_listings, pruned_mailboxes}.
Source
publish(topic : String, envelope : Envelope)

Publish an envelope to all subscribers of a topic. The envelope's to is set to each subscriber's address on delivery.

Source
remove_mailbox(address : String)

Remove a mailbox. Messages in flight are lost.

Source
request(envelope : Envelope, timeout : Time::Span = 30.seconds) : Envelope | Nil

Send an envelope and wait for a reply. Creates a temporary reply mailbox, sets reply_to, and blocks until a response arrives or the timeout expires. The reply mailbox is cleaned up automatically.

Source
resolve_ordering(envelope : Envelope) : Ordering

Resolve ordering: syntactic — service addresses (colon) are sync, agent addresses (no colon) are async.

Source
send(envelope : Envelope)

Send an envelope to its to address. Raises if no mailbox exists, or MailboxFull if the recipient's queue is at capacity.

Source
send?(envelope : Envelope) : Bool

Send, but return false instead of raising when the target mailbox doesn't exist OR is full. Backpressure and missing-address both collapse to "message didn't land" — callers who want to distinguish should use send and catch specific errors.

Source
send_expecting(envelope : Envelope) : String

Send an envelope and register an expectation for a reply on the sender's mailbox. Returns the correlation_id for tracking.

Source
subscribe(topic : String, address : String)

Subscribe an address to a topic.

Source
subscribers(topic : String) : Array(String)

List subscribers for a topic.

Source
subscriptions(address : String) : Array(String)

List topics an address is subscribed to.

Source
system_topics

When true, every successful send fans out a metadata copy to the sys.message.sent topic. Reactive infrastructure (like AIX's tmux bridge) subscribes to it and gets push-quality wake notifications without polling GET /events. Default OFF — enable via ARCANA_SYSTEM_TOPICS=1 in the daemon, or set directly here. Fanout cost: one extra Mailbox#deliver per subscriber per send.

Source
system_topics=(system_topics : Bool)

When true, every successful send fans out a metadata copy to the sys.message.sent topic. Reactive infrastructure (like AIX's tmux bridge) subscribes to it and gets push-quality wake notifications without polling GET /events. Default OFF — enable via ARCANA_SYSTEM_TOPICS=1 in the daemon, or set directly here. Fanout cost: one extra Mailbox#deliver per subscriber per send.

Source
unsubscribe(topic : String, address : String)

Unsubscribe an address from a topic.

Source

Nested types