bus
The Bus wire format and the broker client.
bus::wire(always built):BusMessage,parse,parse_strict,parse_lenientand the size caps. A message is----fencedkey: valueheaders (sorted) plus an optional body. The format is a frozen contract with every running peer.bus::client(the defaultclientfeature):Connection, one registered WebSocket to the broker (noded);SupervisedClient, the same under a reconnect supervisor whose incoming stream survives broker restarts and which replays every recorded topic subscription before reportingConnected;IncomingCommand; andnoded_url(), which finds the broker (MIXOS_NODED_URL, thennode.conf.mixviaMIXOS_NODE_CONFIG,$MIXOS_ETC,~/.config/mixosor/etc/mixos, thenws://127.0.0.1:4200/ws).
A wire-only consumer depends on bus with default-features = false and
pulls in nothing but serde_json.
The source is libs/bus.
Starting without a broker
Resident GUI hosts can use SupervisedClient::connect_options(name, url).start()
inside their existing Tokio runtime to return immediately while the broker is
unavailable. The same supervisor handles initial dial, registration, topic
replay and later reconnects. Until first success the state is Connecting,
generation is zero and outbound work fails fast; first complete
registration and replay publishes generation one. The single incoming
receiver remains usable across outages. connect().await retains its
five-attempt initial budget for callers that need bounded startup. Neither
API queues outbound work.
Bounded incoming lanes
bounded_incoming(capacity) drops overflow silently, reported only as an
Overflow marker. bounded_incoming_rpc(capacity) opts into refusal: an
id-bearing request with no topic header that the full lane cannot retain
gets an rc 10 BUSY reply on the same connection it arrived on, so the
caller can retry. Topic deliveries and id-less frames remain lossy. If that
refusal cannot be written within 100 ms, the old connection gets a
best-effort close within a further 100 ms, and the supervisor detaches its
handle before redialling; the physical close may follow later if another
holder still references it. No reply is ever sent through the replacement
connection.
Registration rejections
With fatal_on_registration_rejection(true), registration_rejection()
retains the broker's exact return code and diagnostic before publishing
Fatal, for both initial registration and reconnect. Sampling it is
non-consuming; transport failure has no refusal diagnostic. Close,
shutdown, deregister and drop are safe before a first socket exists and
fence late registration publication. These are client lifecycle APIs; the
frozen ABP wire format is unchanged.
A broker rejection is typed: RegistrationRejected keeps the public rc
and message fields. Field reads and the tuple accessors stay
source-compatible, but old two-field struct literals no longer compile
(construct with RegistrationRejected::new). It adds kind(), which
decodes the structured rejection body
({"schema": "noded.registration-rejection.v1", "error_code": "NAME_TAKEN", "message": …},
sent alongside the unchanged rc=10 and error header).
RegistrationRejectionKind::NameTaken is produced only from that body; an
admission refusal, a legacy text-only broker, or a missing, malformed,
oversized or unrecognised body is Unknown: classification never
string-matches the wording. ClientError::registration_rejection_typed()
and SupervisedError::registration_rejection_typed() expose the typed
value, and the (rc, message) tuple accessors remain for compatibility.
Tests
cargo test -p bus. The integration test drives a stub broker in process;
no noded is needed.