Realtime uses the binary Centrifugo Protobuf WebSocket protocol (always included; the Cargo realtime feature flag is a no-op stub). Helpers return TypedSubscription<T>. Prefer recv_result().await so a failed feed cannot look like a clean end of stream.
The SDK negotiates the centrifuge-protobuf WebSocket subprotocol. Both Centrifugo control frames
and :proto channel publications are binary Protobuf. ConnectRPC's optional JSON wire mode is
unrelated and never changes realtime framing.
Managed order-book / market-overview subscriptions expose updates().recv().await plus set_on_error(...) and err().
Contract
use polyester::Error;
let mut sub = client
.orders
.subscribe(client.default_account_id.as_deref())
.await?; // handshake done
sub.set_on_error(|error| eprintln!("realtime interruption: {error}"));
while let Some(order) = sub.recv_result().await? {
println!("{} {}", order.status, order.order_id);
}- Private channels need API key + Account ID.
- Overflow is fail-closed (
Error::QueueOverflow). - Reconnect uses capped exponential backoff with per-subscription jitter.
- Hyphenated channel segments (e.g.
api-keys) must be signed with RFC 3986-preserving query encoding (built into the SDK).
What you can stream
| Stream | Method | Auth | Consume |
|---|---|---|---|
| Public trades | market_data.subscribe_trades | public | recv_result() |
| Candles | market_data.subscribe_candles | public | recv_result() |
| Order book (managed) | orderbook.create_subscription | public | updates() |
| Market overview (managed) | market_overview.create_subscription | public | updates() |
| Heatmap | heatmap.subscribe_live | public | recv_result() |
| Lifecycle | lifecycle.subscribe_open_flows / subscribe_flow_detail | public or private | recv_result() |
| Zipped-asset supply | zipper.subscribe_zipped_asset_supply | public | recv_result() |
| Profile identity | auth.profile.subscribe_identity | public | recv_result() |
| Orders / triggers / balances / trades | matching subscribe helpers | private | recv_result() |
| Transfers | transfers.subscribe | private | recv_result() |
| API keys | api_keys.subscribe | private | recv_result() |
| API/subaccount policies | policies.subscribe_*_policies | private | recv_result() |
| Subaccounts / address book | matching subscribe helpers | private | recv_result() |
Error::PermissionDenied is non-transient; inspect code, context, and endpoint, then update the key policy before reconnecting.Public trades
client.wait_for_catalogs().await?;
let mut sub = client.market_data.subscribe_trades("BTC-USDT").await?;
while let Some(trade) = sub.recv_result().await? {
println!("{trade:?}");
break;
}Snapshot-then-stream
Managed order-book subscriptions snapshot via REST, then stream deltas with gap recovery. Prefer these helpers over wiring raw channels yourself. The create call returns only after both the WebSocket handshake and initial snapshot succeed.
Market overview uses the same managed pattern. For entity streams (orders, balances, trades, transfers, triggers), events are not a complete starting snapshot: subscribe first, fetch the corresponding list/snapshot, buffer events during the read, then reconcile by entity ID plus version/timestamp. After reconnect, re-read authoritative state before declaring the bot healthy.
Typed subscriptions do not recover missed publications across reconnect (no Centrifugo resume
cursor). Poll sub.take_resubscribed() (or watch sub.resubscribes()) after each receive and
treat a latch as a gap: rebuild from REST before trusting the stream again. A WebSocket read
timeout is treated as connection death so half-open sockets reconnect instead of freezing.
Never quote from raw order-book deltas. Use orderbook.create_subscription, which detects gaps
and refreshes its snapshot.