Streaming

Realtime uses the binary Centrifugo Protobuf WebSocket protocol, and your order updates use the private WebSocket. Service subscribe helpers return *realtime.Subscription[T]. Consume with Messages(); Go has no Recv().

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 return dedicated types with Updates(), SetOnError(...), and Err().

Contract

sub, err := client.Balances.Subscribe(ctx, accountID)
if err != nil { log.Fatal(err) } // handshake already completed
defer sub.Close()
sub.SetOnError(func(err error) {
    log.Printf("realtime interruption: %v", err)
})

for balance := range sub.Messages() {
    fmt.Println(balance.AssetID, balance.Available)
}
if err := sub.Err(); err != nil {
    // includes *errors.QueueOverflowError when the consumer lagged
}

• Subscribe waits for handshake (and private token fetch) before returning. Orders.Subscribe is the exception: it returns at once and reports the connection as events.
• Private channels need API key + Account ID. Hyphenated segments (api-keys, api-policies) are valid; the SDK's RFC 3986-preserving query encoding leaves - unescaped.
• Pass the Account ID string, or nil for the client default. An integer addresses the account it names (encoded as its public ID); 0, negative and non-integer values return *errors.ValidationError before connecting (details on the Realtime reference page).
• Overflow is fail-closed (*errors.QueueOverflowError).
• Reconnect uses capped exponential backoff with per-subscription jitter.

What you can stream

Every stream below is consumed with Messages(), except the managed order book and market overview, which use Updates().

• Public trades (public): MarketData.SubscribeTrades
• Candles (public): MarketData.SubscribeCandles
• Order book (managed) (public): Orderbook.CreateSubscription
• Market overview (managed) (public): MarketOverview.CreateSubscription
• Heatmap (public): Heatmap.SubscribeLive
• Open flows (public or private): Lifecycle.SubscribeOpenFlows
• One flow (public): Lifecycle.SubscribeFlowDetail
• Zipped-asset supply (public): Zipper.SubscribeZippedAssetSupply
• Profile identity (public): Auth.Profile.SubscribeIdentity
• Your orders (private): Orders.Subscribe
• Triggers / trades (private): matching Subscribe helpers
• Balances (private): Balances.Subscribe
• Transfers (private): Transfers.Subscribe
• API keys (private): APIKeys.Subscribe
• API/subaccount policies (private): Policies.Subscribe*Policies
• Subaccounts / address book (private): matching subscribe helpers

Public trades

symbol := "BTC-USDT"
sub, err := client.MarketData.SubscribeTrades(ctx, &symbol, nil)
if err != nil { log.Fatal(err) }
defer sub.Close()
for trade := range sub.Messages() {
    fmt.Println(trade)
    break
}

Your orders

Orders.Subscribe streams an account's orders over the private WebSocket. It yields realtime.OrderStreamEvent values, so switch on the event type:

import "github.com/Fabric-Labs/polyester-sdk-go/realtime"

sub, err := client.Orders.Subscribe(ctx, accountID)
if err != nil { log.Fatal(err) }
defer sub.Close()
for event := range sub.Messages() {
    switch e := event.(type) {
    case realtime.OrderSnapshot:
        fmt.Println("live orders", len(e.Orders))
    case realtime.OrderUnit:
        for _, order := range e.Orders {
            fmt.Println(order.Status, order.OrderID)
        }
    case realtime.OrderStreamError:
        log.Printf("order stream error: %v", e.Err)
    }
}
if err := sub.Err(); err != nil {
    log.Printf("order stream ended: %v", err)
}

An OrderSnapshot is the account's full live view: replace local state with it. An OrderUnit is one change: its Orders are complete, so replace each by OrderID and keep the higher Version. OrderReady means the stream reflects every acknowledgement given before the subscribe; wait for it before trusting that an order missing from the stream does not exist. The stream reconnects and resumes by itself. See Orders for every event.

Snapshot-then-stream

Managed order-book subscriptions snapshot via REST, then stream deltas. Sequence gaps trigger a REST snapshot refresh. The create call returns only after the WebSocket handshake and initial snapshot succeed. Prefer these helpers over wiring raw channels yourself.

Market overview uses the same managed pattern. For entity streams (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.

Centrifugo subscriptions do not recover missed publications across reconnect (no resume cursor). Poll sub.TakeResubscribed() (or watch sub.Resubscribes()) after each message 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.CreateSubscription, which detects gaps and refreshes its snapshot.

Overflow

Reference: Realtime. Session model: WebSocket session model.

Related Topics