Messaging

Subscribe to a segment, exchange byte payloads, and understand what publish completion means.

Before you start: use the client, the unconnected channel, and ctx from Connection and lifecycle. Your credentials must grant read/write access to chat.

A channel connection automatically joins default. Steps 1 and 2 use an additional named segment, chat, so they explicitly subscribe. To use default directly, skip to step 3.

chat, err := channel.Segment("chat")

if err != nil {
    log.Fatal(err)
}

stopListening := chat.OnMessage(func(payload []byte, metadata celeris.MessageMetadata) {
    fmt.Println(metadata.TokenReference, string(payload))
    fmt.Println("Message ID:", metadata.MessageID)
})

membership, err := chat.Subscribe()

if err != nil {
    log.Fatal(err)
}

if err := channel.Connect(ctx); err != nil {
    log.Fatal(err)
}

OnMessage registers a local listener. Subscribe records interest in the segment and joins it on the server. A call to only OnMessage does not join chat. The default segment is different because the channel connection already joins it.

The listener receives the payload first, then the metadata. metadata.TokenReference is the reference claim in the token of the sender. Presence events and presence lists show the same value as TokenReference. The payload is the SDK's own copy, and every listener of that delivery shares it. Treat the payload as read-only. Copy it before you keep it. Multiple handles for the same segment share one listener set and one interest count.

A channel is one WebSocket connection. A segment is not a connection. When you connect a channel, the SDK opens a new connection. The SDK also opens a new connection for a channel reference that a different channel uses. All segments of a channel use the connection of that channel, and channel.Segment() does not open a connection. To divide your messages into groups, use more segments on one channel, not more channels.

2. Publish bytes

if err := chat.Publish(ctx, celeris.TextPayload("Hello from this connection")); err != nil {
    log.Println("Publish failed:", err)
}

Run a second connected subscriber to observe receipt. By default, a connection does not receive its own publish. Your server can opt into echo with AllowEcho: true when it signs credentials.

A nil result from Publish means only that the local socket accepted the bytes. A server permission denial or size rejection can arrive later through channel.Events().OnError. There is no per-publish receipt.

A publish does not join the connection to a segment. To receive messages from a named segment, call Subscribe(). A listener alone does not receive messages.

This rule also applies to echo. With AllowEcho: true, the connection receives its own publish only while it holds a subscription to that segment.

Segment permissions

The server applies the permissions in your credentials to each segment. Read access is necessary to subscribe and to use presence. Write access is necessary to publish. For the full rules, see Permissions.

On a read-only segment, the connection can subscribe and receive messages, but the server refuses a publish.
On a write-only segment, the connection can publish, but the server refuses a subscription.
On a segment that is not in the claims, the server refuses Subscribe, Publish and presence.

Each refusal is a PermissionDeniedError that names the command and the segment. It goes to channel.Events().OnError, but PresenceList returns a refused presence query as a *celeris.ServerError. The connection stays open.

For examples of write-only access, see When to Use Write-Only Permissions.

Listen to all segments of the channel

A segment listener receives the messages of one segment. channel.Events().OnMessage receives all messages that the connection gets, from all segments. These segments are default and the segments that you subscribe to. The listener also receives messages from segments that have no segment listener. If your code sends each message to the correct handler, use this listener. One channel listener can replace one listener for each segment.

removeChannelListener := channel.Events().OnMessage(func(payload []byte, metadata celeris.MessageMetadata) {
    fmt.Println(metadata.SegmentID, string(payload))
})

The channel listener gets the same payload and metadata as the segment listeners, after the segment listeners. It gets each message ID one time. All listeners of a message get the same bytes. Do not change these bytes.

OnMessage returns a function that removes only this channel listener. When you call removeChannelListener():

The other channel listeners continue to receive messages.
The segment listeners continue to receive messages.
Your subscriptions do not change, and the SDK does not send a message to the server.

When you call the function again, it has no effect.

Optional message identifiers

if err := chat.PublishWithMessageID(ctx, celeris.TextPayload("Order 1042 shipped"), "order-1042-shipped"); err != nil {
    log.Println("Publish failed:", err)
}

Publish generates an ID: msg__go_v, the SDK version, two underscores, and 32 random lowercase hex digits, for example msg__go_v1.0.0__4f1c9a0b7d2e43f6a8b5c1d0e9f27364. A generated ID has only letters, digits, dots, and underscores. The double underscores keep it different from the IDs that the server generates (msg_{node}_…). A receiver can then recognize the publish if the SDK resends it. A reused ID does not guarantee an exactly-once publish. The client filters repeated IDs only within a bounded window (DeduplicationWindowSize, 1024 IDs per channel by default). Use your own durable idempotency for business operations. Your own ID must have at least one character and must not contain CR or LF. The SDK sends it as you give it.

Size and disconnected behavior

The SDK rejects an encoded command larger than 2 MiB with ErrConfiguration and does not write it. Framing, identifiers, and payload together count toward that bound.

The server enforces a lower limit set by your plan. That limit also counts the full encoded command: the framing, the identifiers, and the payload. See Limits for current values. A plan-limit rejection arrives asynchronously and still counts toward usage. Keep the payload below the limit, with space for the framing and the identifiers. Do not use a rejection as a size probe.

If the connection closes, a publish that the SDK did not give to the socket stays in the publish queue. While the channel is reconnecting, a new publish also goes into the queue. After the reconnect, the SDK first sends your message subscriptions again, then your presence subscriptions. Then it sends the queued publishes in the sequence of the calls. Publish returns when the SDK sends the publish.

The SDK does not send a publish again if it gave that publish to the socket before the connection closed. Only rate-limit recovery sends a publish again. When recovery stops in failed, the queued publishes return the error that stopped recovery. When you close the channel, they return ErrCancelled. A publish before the first connect, during the first connect, or after failed returns ErrNotConnected. The queue is in memory only, and there is no durable offline queue.

Publishes wait for room in the writer. ErrBackpressure means that the publish queue (PublishQueueSize, 64 by default) is full. This limit also applies while the channel is reconnecting. Wait until some publishes go out, then retry. The publishes that return to the queue after a connection loss always stay in the queue, even if the queue then holds more than PublishQueueSize publishes.

ErrDeliveryUnknown means that the socket failed, or your context ended, during the write. Acceptance is then uncertain. A blind retry can duplicate the action. If you cancel ctx before the write, the SDK withdraws the publish with ErrCancelled, or with ErrTimeout for a deadline.

When the server rate-limits the connection, its RateLimitError does not say which command it dropped. The SDK then pauses all sends for at least one second. After the pause, the SDK resends what it sent in the last two seconds:

Subscription changes go first.
Up to the last 64 publishes follow, each at most once and with its original ID.

A receiver drops a resent copy that arrived earlier. Resends count toward usage.

After eight rate limits in a row, the SDK treats the limit as a used-up quota and stops the resends. It retries the dropped subscriptions after one minute. That delay doubles each time, to at most one hour, until commands go through without a limit again. If rate limits continue, publish less often.

3. Use the default segment deliberately

lobby := channel.DefaultSegment() // the same as channel.Segment("default")

stopLobby := lobby.OnMessage(func(payload []byte, metadata celeris.MessageMetadata) {
    fmt.Println(metadata.TokenReference, string(payload))
})

if err := lobby.Publish(ctx, celeris.TextPayload("Hello, default")); err != nil {
    log.Println("Publish failed:", err)
}

You are already a member of default when the channel connects. This handle accesses that membership. It does not create a new socket or require Subscribe(). Use it when one stream per channel is enough.

Automatic membership does not bypass authorization. Your credentials still need read permission on default to receive and write permission to publish. default is a stream, not a wildcard. It does not receive messages published to chat or to other named segments. There is no remote unsubscribe for the default segment. Automatic membership does not subscribe to presence.

4. Clean up subscriptions and listeners

stopListening()
membership.Cancel()
channel.Close()

Listener removal and subscription cancellation are different actions. When one caller cancels a subscription, the subscriptions of other callers stay. On a segment that is not default, the SDK tells the server to unsubscribe when you cancel the last message subscription. A presence subscription does not keep the segment. To remove a channel listener, call the function that Events().OnMessage returns, for example removeChannelListener().

The SDK restores held interests after a reconnect. Then it sends the queued publishes. It does not send again a publish that it gave to the socket before the connection closed. Read Reconnection and Recovery.

Next: Payload formats.

Connection and Lifecycle Payload Formats