Go Client API

go get github.com/useceleris/sdk-go-client

The module requires Go 1.27 or newer, and the package name is celeris. The runtime dependency is github.com/coder/websocket. Clients, channels, segments and subscriptions are safe for concurrent use.

NewClient()

func NewClient(options ClientOptions) (*Client, error)

type ClientOptions struct {
    CredentialProvider       CredentialProvider
    BaseURL                  string
    AllowInsecureLoopback    bool
    ConnectTimeout           time.Duration
    ReconnectTimeout         time.Duration
    PresenceQueryTimeout     time.Duration
    PublishQueueSize         int
    DeduplicationWindowSize  int
    MaximumReconnectAttempts int
}

NewClient performs no network work. A zero option means its default. NewClient returns ErrConfiguration for an omitted provider, a negative size, a timeout that is not 0 or from 1 ms to 15 minutes, and a MaximumReconnectAttempts value that is not from 0 to 100. BaseURL is the realtime socket endpoint, not your HTTP credential endpoint. It must be wss://, or ws:// to localhost or a loopback address with AllowInsecureLoopback, and it must carry no credentials, query string or fragment. The SDK dials with its own HTTP client, which honors the proxy environment variables. Changes to http.DefaultTransport do not affect it.

Client

func (client *Client) Channel(reference string) (*Channel, error)

Channel is side-effect free and returns a new connection handle on every call. A reference is 1 to 255 characters of [A-Za-z0-9_-]. Anything else returns ErrConfiguration.

Channel

func (channel *Channel) State() ChannelState
func (channel *Channel) Connect(ctx context.Context) error
func (channel *Channel) Close()
func (channel *Channel) Segment(segmentID string) (*Segment, error)
func (channel *Channel) DefaultSegment() *Segment // Segment("default")
func (channel *Channel) Events() ChannelEventHandler

type ChannelState string

const (
    StateIdle         ChannelState = "idle"
    StateConnecting   ChannelState = "connecting"
    StateConnected    ChannelState = "connected"
    StateReconnecting ChannelState = "reconnecting"
    StateFailed       ChannelState = "failed"
    StateClosing      ChannelState = "closing"
    StateClosed       ChannelState = "closed"
)

You can recover a failed channel when you call Connect again. You cannot recover a closed channel. The SDK does not automatically retry an initial connection failure. See Reconnection and Recovery for triggers, deadlines, retry budgets and restored subscriptions.

A successful Connect joins default. DefaultSegment().OnMessage(listener) receives its messages without Subscribe(). Read and write permissions still apply, and default membership includes no other segment and no presence subscription. The automatic join also happens on reconnect.

Segment

func (segment *Segment) ID() string
func (segment *Segment) Subscribe() (*Subscription, error)
func (segment *Segment) SubscribePresence() (*Subscription, error)
func (segment *Segment) OnMessage(listener func(payload []byte, metadata MessageMetadata)) (remove func())
func (segment *Segment) OnPresence(listener func(PresenceEvent)) (remove func())
func (segment *Segment) Publish(ctx context.Context, payload []byte) error
func (segment *Segment) PublishWithMessageID(ctx context.Context, payload []byte, messageID string) error
func (segment *Segment) PresenceList(ctx context.Context, page, perPage int32) (PresencePage, error)

func (subscription *Subscription) Cancel() // idempotent

Payload helpers

func TextPayload(text string) []byte
func JSONPayload(value any) ([]byte, error)
func ReadText(payload []byte) (string, error)
func ReadJSON[T any](payload []byte) (T, error) // decodes into T; does not validate

func NewPayloadCodec[T any](encode func(T) ([]byte, error), decode func([]byte) (T, error)) (PayloadCodec[T], error)
func (codec PayloadCodec[T]) EncodePayload(value T) ([]byte, error)
func (codec PayloadCodec[T]) ReadPayload(payload []byte) (T, error)

TextPayload replaces invalid UTF-8 with U+FFFD. JSONPayload writes compact JSON with encoding/json. It does not escape HTML and does not add a final newline. It returns ErrConfiguration for NaN, infinities, channels, functions, complex numbers and cyclic structures. ReadText and ReadJSON return ErrConfiguration for invalid UTF-8, invalid JSON, or JSON that does not fit T. NewPayloadCodec returns ErrConfiguration for a nil function. The SDK returns errors from your own encode or decode unchanged. No size check happens here: Publish enforces the encoded-command bound.

Types

type MessageMetadata struct {
    TokenReference string
    SegmentID      string
    MessageID      string // always present: the publisher's, or one the server assigned
    Timestamp      int64  // Unix milliseconds
}

type PresenceEvent struct {
    SegmentID      string
    TokenReference string
    ConnectionID   string
    Joined         bool // false: the connection left
    Timestamp      int64
}

type PresencePage struct {
    SegmentID   string
    Total       int32
    PerPage     int32
    CurrentPage int32
    From        int32 // From > To past the last page
    To          int32
    Connections []PresenceConnection
}

type PresenceConnection struct {
    TokenReference string
    ConnectionID   string
    Timestamp      int64
}

type ServerNotice struct {
    Timestamp int64
    Payload   []byte
}

type RecoveryEvent struct {
    RetryIndex         int
    PossibleGaps       bool // always true
    PossibleDuplicates bool // always true
}

type CredentialRequest struct {
    ChannelReference string
    Reconnect        bool          // false for the attempt Connect starts
    DisconnectedAt   time.Time     // zero unless Reconnect
    ReplayLookback   time.Duration // the outage plus 5 s; zero unless Reconnect
}

type Credentials struct {
    Payload   string `json:"payload"`
    Signature string `json:"signature"`
}

type CredentialProvider func(ctx context.Context, request CredentialRequest) (Credentials, error)

Presence figures are signed 32-bit values as received. Timestamps are signed 64-bit Unix milliseconds. ReplayLookback is whole milliseconds, at most 4294967295 ms. Credentials implements fmt.Stringer, fmt.Formatter and slog.LogValuer, so it prints as redacted with every verb when you format it directly. fmt prints a struct's unexported fields field by field, so do not keep credentials in a struct you print. A provider must honor ctx and return non-empty, valid UTF-8 values. Empty or invalid values return ErrConfiguration. The SDK reports any error the provider returns, or a panic, as ErrTransport.

ChannelEventHandler

func (handler ChannelEventHandler) OnStateChange(listener func(ChannelState)) (remove func())
func (handler ChannelEventHandler) OnRecovery(listener func(RecoveryEvent)) (remove func())
func (handler ChannelEventHandler) OnNotice(listener func(ServerNotice)) (remove func())
func (handler ChannelEventHandler) OnError(listener func(error)) (remove func())
func (handler ChannelEventHandler) OnMessage(listener func(payload []byte, metadata MessageMetadata)) (remove func())

Each registration returns a function that removes it. Listeners run one at a time, in the order events happened, never under an SDK lock, and with no queue on your side. A slow listener delays delivery and does not grow a backlog. A listener can call back into the channel. The events its calls cause follow once it returns. The SDK recovers a listener that panics and reports it through OnError. The SDK does not recover a panic in an error listener. A recovery event follows the connected state change. It does not signal replay completion.

OnMessage receives all messages from all segments of the connection. It gets each message after the segment listeners, and it gets each message ID one time. All listeners of a message get the same payload. Do not change the payload.

The function that OnMessage returns removes only this channel listener. The other channel listeners and the segment listeners continue to receive messages. The subscriptions do not change, and the SDK does not send a message to the server. When you call it again, it has no effect.

OnError receives failures that no caller waits for: a *ServerError, or an *Error such as an undecodable frame, a reconnect that gave up, or a listener that panicked. OnNotice delivers raw server prose, such as subscription acknowledgments and refusals. It carries no segment id and no correlation id. Do not parse it.

Errors

type ErrorCode string // a sentinel: errors.Is(err, ErrTimeout)

type Error struct {
    Code    ErrorCode
    Message string
    Field   string // set only for ErrProtocol
    Offset  int    // zero-based byte offset, set only for ErrProtocol
}

const (
    ErrConfiguration       ErrorCode = "Configuration"
    ErrTimeout             ErrorCode = "Timeout"
    ErrCancelled           ErrorCode = "Cancelled"
    ErrTransport           ErrorCode = "Transport"
    ErrNotConnected        ErrorCode = "NotConnected"
    ErrBackpressure        ErrorCode = "Backpressure"
    ErrOperationInProgress ErrorCode = "OperationInProgress"
    ErrDeliveryUnknown     ErrorCode = "DeliveryUnknown"
    ErrProtocol            ErrorCode = "ProtocolError"
)

Every error the SDK detects is an *Error. Its message says what failed and which rule or limit it broke, such as Invalid channel reference. Must not be empty.. The message never carries your input, credentials or server bytes, and it never wraps a cause that holds them. errors.Is(err, celeris.ErrTimeout) matches by code. Errors the server sends are a *ServerError instead. Never match on message text.

ErrProtocol means the SDK could not decode one message. The SDK drops that message and reports the error, and the connection stays up. The SDK silently skips a command that this SDK version does not recognize, so a newer server cannot break a deployed client.

There is no authentication code. The WebSocket handshake status is not exposed to the SDK, so it reports a refused handshake as ErrTransport and does not guess.

ServerError

type ServerErrorType string

const (
    ParserError           ServerErrorType = "ParserError"
    SendError             ServerErrorType = "SendError"
    PermissionDeniedError ServerErrorType = "PermissionDeniedError"
    RateLimitError        ServerErrorType = "RateLimitError"
    MessageSizeLimitError ServerErrorType = "MessageSizeLimitError"
    InternalError         ServerErrorType = "InternalError"
)

type ServerError struct {
    Type     ServerErrorType // one of the above, or a type a newer server adds
    SubType  string          // the command it answers, e.g. "PUB"; "" when none
    Message  string          // the server's own text, unchanged
    Resource any             // what that command names: nil, string, int32, int64 or []any
}

A ServerError is an error the server sent, with every field as sent, and the connection stays up. It has no Code: match it with errors.AsType. It arrives through OnError, after the call that caused it, except for a presence query error, which PresenceList returns instead. Keep a default branch for types a newer server adds. Check Resource's type before you use it.

Limits

The SDK does not check the size of received messages. They are already in memory when they arrive, so the SDK processes whatever the server sends. A publish rejected for your plan's size limit still counts toward your usage. A local encoded-command rejection sends nothing.

What this SDK will not do

The SDK does not do these tasks:

• Confirm delivery. There are no receipts anywhere in the protocol.
• Keep a durable offline queue. Publishes wait in an in-memory queue only while the channel is reconnecting. The SDK resends publishes only after a rate limit, once each, with their original ID.
• Provide durable history or global ordering. Replay is a bounded window, not a cursor.
• Sign anything. The signer lives in sdk-go-server.