Messaging

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

Before you start: use the client and unconnected channel from Connection and lifecycle. Your credentials must grant read and 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.

from useceleris_client import MessageMetadata, read_text, text_payload

chat = channel.segment("chat")


def receive(payload: bytes, metadata: MessageMetadata) -> None:
    print(metadata.token_reference, read_text(payload))
    print("Message ID:", metadata.message_id)


stop_listening = chat.on_message(receive)
membership = chat.subscribe()
await channel.connect()

on_message() registers a local listener. subscribe() records interest in the segment and joins it on the server. A listener registration alone does not join chat. The default segment is different because the channel connection already joins it.

The listener receives the payload bytes first, then the metadata. metadata.token_reference is the reference claim in the token of the sender. Presence events and presence lists show the same value as token_reference. Every Segment object for the same segment shares 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

await chat.publish(text_payload("Hello from this connection"))

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 "allow_echo": True when it signs credentials.

A completed publish means only that the local socket accepted the bytes. A server permission denial or size rejection can arrive later through channel.events().on_error. 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 "allow_echo": 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().on_error(), but a refused presence query raises a ServerError from presence_list(). 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().on_message() 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.

def show_any(payload: bytes, metadata: MessageMetadata) -> None:
    print(metadata.segment_id, read_text(payload))


remove_channel_listener = channel.events().on_message(show_any)

The channel listener gets the same payload and metadata as the segment listeners, after the segment listeners. It gets each message ID one time.

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

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

import uuid

await chat.publish(
    text_payload("An update with an application-supplied ID"),
    message_id=str(uuid.uuid4()),
)

If you omit the identifier, the SDK generates one: msg__py_v, the SDK version, two underscores, and 32 random lowercase hex digits, for example msg__py_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 in a window of recent IDs (deduplication_window_size, 1024 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 a ConfigurationError 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. When the queue is full, the publish raises Backpressure. The publish_queue_size option sets the queue size, 64 publishes by default. 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. Each 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 raise the error that stopped recovery. When you close the channel, they raise Cancelled. A publish before the first connect, during the first connect, or after failed raises NotConnected. The queue is in memory only, and there is no durable offline queue.

A DeliveryUnknown error means that acceptance is uncertain. A blind retry can duplicate the action. To withdraw an unsent publish, cancel the task that awaits publish().

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.default_segment()  # the same as channel.segment("default")


def show(payload: bytes, metadata: MessageMetadata) -> None:
    print(metadata.token_reference, read_text(payload))


stop_lobby = lobby.on_message(show)
await lobby.publish(text_payload("Hello, default"))

You are already a member of default when the channel connects. This object accesses that membership. It does not open 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

stop_listening()
membership.cancel()
await 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().on_message() returns, for example remove_channel_listener().

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