Messaging

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

Connecting to a channel automatically joins default. Steps 1 and 2 use an additional named segment, chat, so they subscribe explicitly. For messaging directly on default, see step 3.

1. Listen and subscribe

Register interest before connecting:

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. Registering a listener alone does not join chat; default is different because connecting already joins it.

The listener receives the payload bytes first, then the metadata. All segments of the channel share one socket, and every Segment object for the same segment shares one listener set and one interest count.

2. Publish bytes

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

Run a second connected subscriber to watch it arrive. By default a connection does not receive its own publishes; your server can opt in with "allow_echo": True when signing credentials.

When publish() returns, only the local socket has accepted the bytes. A server permission denial or size rejection can arrive later through channel.events().on_error. There is no per-publish receipt.

Publishing also joins the segment server-side. It does not grant read permission: a write-only member receives nothing.

Optional message identifiers

import uuid

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

If omitted, the SDK generates a random one, so a publish can be recognised if it is resent. Reusing an ID is not an exactly-once strategy: the client filters repeated IDs only within a bounded window, so keep your own durable idempotency for business operations.

Size and disconnected behaviour

The SDK rejects an encoded command larger than 2 MiB before writing it, with a ConfigurationError. Framing, identifiers and payload all count toward that bound.

The server enforces lower plan payload limits: Free 64 KiB, Standard 128 KiB, Pro 512 KiB, Prime 1024 KiB. A plan-limit rejection arrives later and still counts toward usage. Check payload size before sending; do not use rejection as a size probe.

There is no offline queue. A publish while disconnected raises NotConnected, and so does a publish still waiting to be sent when the connection drops. DeliveryUnknown means acceptance is uncertain; blindly retrying may duplicate an action. Cancelling the task awaiting publish() withdraws a publish that has not gone out yet.

When the server rate-limits the connection, its RateLimitError does not say which command it dropped. The SDK pauses sending for at least a second, then resends what it sent in the last two seconds: subscription changes first, then up to the last 64 publishes, each at most once with its original ID, so receivers drop a copy that had already arrived. Resends count toward usage. After eight rate limits in a row the SDK treats the limit as a used-up quota: it stops resending and retries the subscriptions it dropped after a minute, doubling to at most an hour, until commands go through without a limit again. If rate limits keep arriving, 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 once the channel connects. This object accesses that membership; it opens no new socket and needs no 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 other segments. It cannot be left, and automatic membership does not subscribe to presence.

4. Clean up subscriptions and listeners

stop_listening()
membership.cancel()
await channel.close()

Removing a listener and cancelling a subscription are separate actions. One caller cancelling does not remove another caller's subscription. On a named segment, releasing the last message interest leaves the segment, unless a presence interest still holds it. Cancelling presence sends only the presence unsubscribe: message membership survives it, even when no message interest remains.

The SDK restores held interests after reconnecting, but a reconnect never resends publishes. Read Reconnection and Recovery.

Next: Payload formats.