Connection and Lifecycle

Before you start: install useceleris-client (see the overview). You also need an endpoint that signs credentials, as Authentication describes. A trusted backend can sign its own credentials instead. See Server-side usage.

1. Fetch credentials

The SDK has a built-in production endpoint. Supply a credential provider: a coroutine that obtains signed credentials from your application for each connection attempt.

import asyncio
import json
import urllib.request

from useceleris_client import CredentialRequest, Credentials, create_client


def post_json(url: str, body: dict[str, object]) -> dict[str, str]:
    request = urllib.request.Request(
        url,
        data=json.dumps(body).encode(),
        headers={"content-type": "application/json", "authorization": "Bearer ..."},
        method="POST",
    )

    with urllib.request.urlopen(request, timeout=10) as response:
        result: dict[str, str] = json.load(response)

    return result


async def provide_credentials(request: CredentialRequest) -> Credentials:
    body = await asyncio.to_thread(
        post_json,
        "https://your-app.example/api/realtime-credentials",
        {
            "channel_reference": request.channel_reference,
            "replay_lookback_ms": request.replay_lookback_ms,
        },
    )
    return Credentials(payload=body["payload"], signature=body["signature"])


client = create_client(credential_provider=provide_credentials)
channel = client.channel("demo-room")

The provider receives a CredentialRequest with channel_reference and a reason ("initial" or "reconnect"). On recovery it also carries disconnected_at, in Unix milliseconds, and replay_lookback_ms, a duration in milliseconds. Both fields are None on the first attempt. Any HTTP client works. This example uses only the standard library.

Return the payload and signature exactly as your backend signed them. Do not generate, cache, or inspect them. The provider runs on every connection attempt. This includes recovery attempts. If the provider raises an exception, the SDK reports a safe Transport error. The SDK does not forward the original failure details.

When you create a client, a channel, or a segment, the SDK does not open a connection. Use base_url only for local or self-hosted Celeris, not your application's credential endpoint. See the Python client API for the options.

2. Observe state, then connect

from useceleris_client import CelerisConnectionError, ChannelError, ChannelState


def show_state(state: ChannelState) -> None:
    print("Connection:", state)


def show_error(error: ChannelError) -> None:
    print("Error:", error.code, error)


events = channel.events()
stop_state = events.on_state_change(show_state)
stop_errors = events.on_error(show_error)

try:
    await channel.connect()
except CelerisConnectionError as error:
    print("Initial connection failed:", error.code)

The expected states are connecting, then connected. A failed initial attempt raises from connect() and sets failed. The SDK does not also report that failure to on_error, and it does not retry the attempt automatically.

Once connected, the channel is automatically a member of default. Use channel.default_segment() to access that stream. Do not open another connection or call subscribe() just to join it. See default-segment messaging.

A connect call in the connected, connecting, or reconnecting state raises OperationInProgress.

To cancel a connection attempt, cancel the task that awaits it. connect() then raises CancelledError, and the channel state is failed:

connecting = asyncio.create_task(channel.connect())
await asyncio.sleep(0)  # the attempt is under way
connecting.cancel()

try:
    await connecting
except asyncio.CancelledError:
    print(channel.state)  # failed

A task canceled before its first run never starts the attempt, and the channel stays idle. asyncio.wait_for(channel.connect(), timeout=5) works the same way. The connect_timeout_ms option, 15 seconds by default, bounds the credentials and the handshake together. reconnect_timeout_ms bounds a reconnect attempt and defaults to the same value. Each timeout option accepts from 1 ms to 15 minutes (900_000). For a larger value, the SDK raises ConfigurationError.

3. Respond to recovery

Register a recovery listener once when you set up the channel:

from useceleris_client import RecoveryEvent


def recovered(event: RecoveryEvent) -> None:
    print("Recovered after retry", event.retry_index)
    # Reload your application's authoritative state here.


stop_recovery = events.on_recovery(recovered)

retry_index is the count of failed reconnect attempts already used in the current budget. It is 0 when recovery succeeds without an earlier failed attempt in that budget. possible_gaps and possible_duplicates are always True. Reload what matters from your own API. Then reconcile new messages with an application-owned version.

The maximum_reconnect_attempts option sets the budget: 10 failed attempts by default, from 1 to 100. Only Transport and Timeout failures use the budget. Any other failure stops recovery immediately. When recovery stops, on_error receives the last error, and the state changes to failed.

Listeners survive reconnects. Do not register them again on every connected state. Catch errors inside any task that you start from a listener. The SDK isolates a failure inside the listener itself, not inside work that the listener starts.

Replay sends at most the last 100 messages of each segment, for at most 2 minutes. default replays when the channel connects. Other segments replay when the connection joins them with subscribe(). A publish starts no replay. If the reconnect credentials have no replay claim, the missed messages are not available, and the recovery event reports the gap. See Reconcile application data.

See Reconnection and Recovery for triggers, timing, restored interests, and replay. The recovery event is not a replay-complete signal.

4. Release resources

Each event registration returns a function that removes it. Subscription objects have cancel(). The channel owns the socket.

await channel.close()
stop_recovery()
stop_errors()
stop_state()

close() is idempotent and permanent. It has a five-second budget for a graceful close. It cancels recovery and closes the channel's segment objects too. Do not reuse a closed channel. Obtain a new one from the client.

Next: Messaging.