Transport

riverbase_core::transport moves messages between processes: integration messages out of the outbox, realtime updates to connected clients, and RPC-style command and query dispatch without HTTP.

Buses

Type	URL scheme	Feature	Good for
NatsMessageBus	nats://	nats-io	Production default; subject wildcards; worker fan-out
RedisMessageBus	redis://, rediss://	redis	Teams already running Redis
PgMessageBus	postgres://, postgresql://	built in	Local development and small deployments: LISTEN/NOTIFY on the app database

All three implement MessageBus (publish) and StreamBus (publish and subscribe).

Selecting one

[riverbase.bus]
url = "nats://127.0.0.1:4222"     # or redis://…, or postgres://… (LISTEN/NOTIFY)

The scheme picks the backend. The URL falls back to RIVERBASE_BUS_URL, then BUS_URL, then nats://127.0.0.1:4222. DomainRuntime::from_config connects it with connect_stream_bus(&config). Enable the matching feature on riverbase_core.

Using one directly

use riverbase_core::command::MessageBus;
use riverbase_core::transport::NatsMessageBus;

let bus = NatsMessageBus::connect("nats://127.0.0.1:4222").await?;
bus.publish("riverbase.todo.created", serde_json::json!({ "id": id })).await?;

Inside a command, prefer emit_message, which goes through the outbox. Publishing directly from a handler bypasses the transaction.

StreamBus and realtime

StreamBus adds subscribe_values. DomainRuntime::stream_bus is the shared instance that riverbase_http uses for SSE, WebSocket, and RxDB, so an update made on one replica reaches clients connected to another.

RPC dispatch

For command and query calls over a transport rather than HTTP:

Type	Role
TransportEnvelope::Command { cmdkey, payload, aggroot }	A command request
TransportEnvelope::Query { resource, access, request, item_id }	A query request
CommandDispatchActor	Holds Arc<dyn DomainCommandEngine> and executes envelopes
QueryDispatchActor	Holds Arc<dyn DomainQueryEngine> and executes envelopes

Transport is not a policy bypass. Dispatched calls still evaluate the target domain's policy.

Consumers

To consume bus messages as jobs with tracking, use riverbase_task's WorkerRuntime::run_nats_consumer. Consumers must be idempotent, because outbox delivery is at least once.