Real-Time Updates (Channels)

Most of a web app is request and response: the browser asks for a page, your code answers, and nothing else happens until the user clicks again. Channels are for the other case - when the server needs to speak first. A new chat message should appear for everyone in the room without anyone refreshing; a long import should push its progress; a notification should pop the moment it is created. That server-initiated traffic runs over a WebSocket, and Channels are how Proper organizes it.

If you have never used one: a WebSocket is a connection the browser opens once, with an ordinary HTTP request that asks to be upgraded, and then keeps open for as long as the page lives. After that, either side can send a message at any moment, without a request waiting for it. One connection per tab carries everything. With Channels you never touch the connection itself; you write Python classes that decide who may listen to what, and call broadcast() when something happens.

After reading this guide, you will know:

• The three pieces a real-time feature is built from, and how a message travels through them.
• How to define a channel, authenticate the connection, and authorize a subscription.
• How to broadcast - from inside a channel, and from a controller or background task.
• How to use the cable.js client, or skip JavaScript entirely with Turbo Streams.
• How the WebSockets are served, how to run them on several machines, and how to test channels without a server.

1. The shape of a channel

A real-time feature in Proper has three parts:

• The channel - a Python class, the WebSocket equivalent of a controller. It decides who may subscribe, handles messages from the client, and sends data back.
• The cable - the server that holds the WebSocket connections and delivers messages. Channels subscribe to named streams; when any code broadcasts to a stream, the cable sends the message to every subscriber.
• cable.js - the browser client. It opens one WebSocket, manages your subscriptions, reconnects when the connection drops, and hands incoming data to your callbacks.

What connects them is the stream: a plain string like chat_42. A channel calls stream_from("chat_42") to start listening, and any code anywhere calls broadcast("chat_42", data) to deliver to every listener. The channel never addresses a specific browser - it addresses a stream, and the cable does the routing.

Here is the whole loop in miniature. A channel that joins a room's stream and re-broadcasts what it is told:

from ..router import router
from .app_channel import AppChannel


@router.channel()
class ChatChannel(AppChannel):
    def subscribed(self):
        self.stream_from(f"chat_{self.params['room']}")

    def speak(self, data):
        self.broadcast(
            f"chat_{self.params['room']}",
            {"message": data["message"]},
        )

And the browser side - subscribe, render what arrives, send what the user types:

import { cable } from "cable"

cable.connect()
const chat = cable.subscribe(
    "ChatChannel",
    { room: "general" },
    { received(data) { addMessageToDOM(data.message) } }
)

chat.perform("speak", { message: "hello" })

Follow one message through the pieces. Ana types "hello" in room general, where Bo also has the page open:

1. Ana's page called cable.subscribe("ChatChannel", {room: "general"}, ...) when it loaded. The cable created a ChatChannel for her, ran its subscribed(), and the channel started listening to the stream chat_general. Bo's page did the same, so there are two subscriptions listening to that stream, on two connections.
2. Ana's page calls chat.perform("speak", {message: "hello"}). The message travels over her WebSocket to the cable, which finds her ChatChannel and calls its speak() method with {"message": "hello"} as data.
3. speak() calls self.broadcast("chat_general", {...}). The cable looks up who listens to chat_general and writes the message to both connections, Ana's and Bo's.
4. In both browsers, cable.js sees that the message belongs to the ChatChannel subscription and calls its received(data) callback, which adds the line to the page.

Step 3 knows nothing about Ana or Bo. The channel addressed the stream, and the cable found who was listening. Every feature in this guide works this way, and the sections below take the pieces one at a time.

2. Installation

Channels is an addon. Install it with:

$ proper install channels

It creates:

• config/channels.py - CABLE_PATH, CABLE_PORT, and CABLE, the cable backend. It is imported from config/__init__.py.
• channels/app_channel.py - the AppChannel base your own channels inherit from. It is to channels what AppController is to controllers.
• channels/__init__.py - where each channel module is imported, so its @router.channel() decorator runs.
• assets/js/cable.js - the browser client. It is added to the import map as "cable", and assets/js/application.js imports it, so every page can use it.
• tests/channels/ - for your channel tests.

That is all the setup there is. proper run starts the cable in the web process, on CABLE_PORT - PORT + 1 by default, so 2301 in development. All the channels of a page share one WebSocket: a browser tab never opens more than one, however many channels it subscribes to.

3. Defining a channel

A channel is a subclass of AppChannel, registered with the router by the @router.channel() decorator. Use the generator to add one:

$ proper g channel Chat

It creates channels/chat_channel.py and imports it from channels/__init__.py. Fill it in:

from ..router import router
from .app_channel import AppChannel


@router.channel()
class ChatChannel(AppChannel):
    def subscribed(self):
        room = self.params["room"]
        self.stream_from(f"chat_{room}")

    def speak(self, data):
        self.broadcast(f"chat_{self.params['room']}", {
            "message": data["message"],
        })

The channel is registered under its class name - "ChatChannel" - and that is the name clients subscribe with. To register it under another name, pass it to the decorator: @router.channel("chat").

The params are whatever the client passed when subscribing (here, {"room": "general"}), available as self.params for the life of the subscription. The channel name and the params together identify a subscription: a client can subscribe to ChatChannel twice with different rooms, and each one is a separate subscription, with its own channel instance.

Inside any channel method you have:

4. The lifecycle: subscribed and unsubscribed

Two methods bracket a subscription. Override the ones you need:

@router.channel()
class ChatChannel(AppChannel):
    def subscribed(self):
        # Authorize, set up streams, send a welcome.
        self.stream_from(f"chat_{self.params['room']}")
        self.send({"status": "joined"})

    def unsubscribed(self):
        # Clean up anything subscribed() set up outside of streams.
        pass

subscribed() runs once, when the client subscribes. It is where you decide whether the subscription is allowed (see Authorizing versus authenticating). Call stream_from() to start listening, and optionally send() a first message. What you send() here is held back and delivered just before the subscription is confirmed; if you reject(), it is discarded and never reaches the client.

unsubscribed() runs when the client unsubscribes, when the connection closes - the tab was closed, the network dropped, or the server closed it - and when the server stops. Proper removes the channel from all its streams before calling it, so you only need to undo work that lives elsewhere.

5. Authenticating a connection

A WebSocket handshake is an ordinary HTTP request, so it carries the same cookies your controllers see - including the signed session cookie of a logged-in user.

When the auth addon is installed, AppChannel uses that cookie to find the user, and exposes it as current.user, the same way a controller sees it:

from proper import current

from ..router import router
from .app_channel import AppChannel


@router.channel()
class InboxChannel(AppChannel):
    def subscribed(self):
        if not self.authenticated:
            self.reject()
            return
        self.stream_from(f"inbox_{current.user.id}")

current.user is the server-verified user, and self.authenticated is True when there is one. current.user is available in every channel method: subscribed(), your actions, and unsubscribed().

This is the AppChannel the addon creates:

from proper.channels import Channel
from proper.models.base import ProperModel


try:
    from ..models import Session, User
except ImportError:
    # The auth addon is not installed, so channels stay anonymous.
    Session = None


class AppChannel(Channel):
    Session = Session

    def find_user(self, user_id) -> "User | None":
        return User.get_or_none(User.id == user_id)

Two things make it work:

• Session is your app's session model. The first channel a connection subscribes to reads the session cookie, finds the session, and remembers the user's id for the whole connection. The other channels of that connection reuse it, so the cookie is read once per connection. current.auth_session is set only in the subscribed() of that first channel.
• find_user() loads the user by that id. It runs before every other call to the channel - each action, the subscribed() of the later channels, and unsubscribed() - so current.user is always fresh from the database. Change it to fit your app, for example to treat a deactivated user as no user.

Without the auth addon, Session is None, every connection is anonymous, current.user is None, and self.authenticated is False.

5.1 Clients without cookies

A mobile app or a script has no session cookie. The handshake is still an HTTP request, and the channel sees all of it in self.request: the path and query string (request.query), the Authorization header, and the client's address (request.remote_ip, which honors X-Forwarded-For from your proxy when it is one of the TRUSTED_PROXIES; see Deployment). A handshake without a peer address (only the in-memory test server's connect() without details sends none; the cable always does) gets no X-Forwarded-For, and its remote_ip is "". The session lookup is one method, find_session(), that reads the signed cookie by default. Override it in AppChannel to accept a token as well:

class AppChannel(Channel):
    Session = Session

    def find_session(self):
        auth = self.request.headers.get("authorization", "")
        if auth.startswith("Bearer "):
            return Session.find_by_token(auth.removeprefix("Bearer "))
        return super().find_session()

    def find_user(self, user_id) -> "User | None":
        return User.get_or_none(User.id == user_id)

It runs once per connection, when its first channel subscribes; return None to leave the connection anonymous. Prefer the header to a token in the query string, which ends up in proxy logs.

5.2 Authorizing versus authenticating

Authentication answers "who is connected?"; authorization answers "may they subscribe to this?". Do the second in subscribed(), using the first:

def subscribed(self):
    room = Room.get_or_none(Room.id == self.params["room_id"])
    if room is None or not room.has_member(current.user):
        self.reject()
        return
    self.stream_from(f"room_{room.id}")

reject() denies the subscription: the client gets a reject_subscription message, the channel is discarded, and none of its actions can be called. The client never sends a stream name, only params; the channel checks them and picks the stream on the server. So a client cannot listen to a room it does not belong to.

6. Actions: messages from the client

Any public method of a channel can be called by the client as an action:

from proper import current


@router.channel()
class ChatChannel(AppChannel):
    def subscribed(self):
        self.stream_from(f"chat_{self.params['room']}")

    def speak(self, data):
        self.broadcast(
            f"chat_{self.params['room']}",
            {"message": data["message"], "sender": current.user.login},
        )

    def typing(self, data):
        self.broadcast(
            f"chat_{self.params['room']}",
            {"typing": current.user.login},
        )

The client calls chat.perform("speak", {message: "hello"}) and the speak method runs, with the payload as data. An action always receives data (an empty dict if the client sent nothing), so every action method takes that argument.

Actions are plain synchronous Python with a database connection ready, like a controller action. The calls of one connection - its subscriptions, actions and unsubscriptions - run one at a time, in the order they arrived, in a pool of worker threads (four by default, see workers in The cable server). A slow action keeps one of those threads busy, so move long work to a background task that broadcasts when it is done.

These names are refused as actions, with an error frame:

Any other public method is reachable from the client, so keep helpers private with a leading _.

There is one conventional action name: receive. The client's subscription.send(data) is shorthand for perform("receive", data), so if you define a receive(self, data) method it becomes the default handler for that subscription.

6.1 Replying to the caller

perform() returns a promise, and what the action returns is what it resolves with. To tell the caller that something was wrong, raise ActionError with a reason and anything else the page needs:

from proper.channels import ActionError


class ChatChannel(AppChannel):
    def speak(self, data):
        text = data.get("message", "").strip()
        if not text:
            raise ActionError("empty")
        if len(text) > 500:
            raise ActionError("too_long", max=500)
        message = Message.create(room_id=self.params["room"], text=text)
        self.broadcast(f"chat_{self.params['room']}", {"message": text})
        return {"id": message.id}

try {
  const { id } = await chat.perform("speak", { message })
  input.value = ""
} catch (error) {
  showError(error.reason === "too_long" ? `At most ${error.max} characters` : "Try again")
}

The promise rejects with the ActionError's data ({reason: "too_long", max: 500}), with {reason: "error"} when the action raised anything else (the error is logged, the client doesn't learn the details), with {reason: "not_subscribed"}, {reason: "invalid_action"} or {reason: "unknown_action"} when there was nothing to run, and with {reason: "timeout"} when no reply came in 10 seconds (perform(action, data, {timeout}) changes that). An action that returns nothing resolves with null. The connection and its other subscriptions keep working after any of these.

Two things to keep in mind. A timeout means the reply didn't arrive, not that the action didn't run: the server may have finished it, and a reply that comes later is dropped, so don't retry an action that creates or sends something on a timeout without the user's say-so, or have it take an id from the client so a repeat is ignored. And you don't have to wait for the reply: chat.perform("typing") on its own is fine, and a rejection nobody waits for is not reported anywhere.

7. Streams and broadcasting

Streams are the unit of delivery. A channel can stream from as many as it likes, and a broadcast to a stream reaches every channel - on every connection - that is listening to it.

Use send() to answer the one client in front of you (a confirmation, a validation error); use broadcast() to tell everyone listening. The data can be anything Proper can encode as JSON, including a string of HTML.

A broadcast is encoded once and every subscriber receives the same frame; it does not go through send(), even one you override. To tell different users different things, give them different streams, such as f"notices_{user.id}".

7.1 Who should get it

The question "who should see this?" has one answer in Channels: whoever listens to the stream you broadcast to. So the shape of a feature is the shape of its stream names:

# Everyone: one stream, every channel listens to the same name
self.stream_from("announcements")
app.cable.broadcast("announcements", {"text": "Deploying in 5 minutes"})

# A group (a room, a team, a document): one stream per group
self.stream_from(f"chat_{room.id}")
app.cable.broadcast(f"chat_{room.id}", {...})

# One person: one stream per user, named after the verified user, never after params
self.stream_from(f"inbox_{current.user.id}")
app.cable.broadcast(f"inbox_{user.id}", {"unread": 3})

# Something that is happening: one stream per job, until it is done
self.stream_from(f"import_{self.params['job_id']}")
app.cable.broadcast(f"import_{job.id}", {"progress": 80})

A channel can listen to several at once - a NotificationsChannel might stream from announcements and inbox_{user.id} - and the page gets both through the same received() callback. Put a field in the data, such as {"kind": "unread", ...}, when the page needs to tell them apart.

7.2 Naming streams

Stream names are arbitrary strings. The convention is a descriptive prefix with a dynamic suffix:

self.stream_from(f"chat_{room_id}")
self.stream_from(f"inbox_{current.user.id}")
self.stream_from(f"document_{doc_id}_edits")

The name is the contract between the channel that listens and the code that broadcasts, and both sides have to spell it the same way. Nothing checks that they agree: a typo on one side delivers to no one, without an error. A small function that builds the name, used on both sides, avoids that:

def chat_stream(room_id):
    return f"chat_{room_id}"

7.3 Broadcasting from a controller or a task

The most common broadcast does not come from a channel at all - it comes from an ordinary request, or a background task, that just changed something open pages should see. Any code with the app can broadcast through app.cable:

from ..models import Message
from ..router import router
from .app_controller import AppController


@router.resource("rooms/:room_id/messages")
class MessageController(AppController):
    def create(self):
        room_id = self.params["room_id"]
        message = Message.create(room_id=room_id, text=self.params["text"])

        # Push to everyone watching this room,
        # then answer this request normally.
        self.app.cable.broadcast(f"chat_{room_id}", {
            "message": message.text,
            "id": message.id,
        })

        self.response.redirect_to("Message.index", room_id=room_id)

A controller reaches the cable through self.app.cable; a background task imports the app:

from ..main import app


@app.queue.task()
def notify_user(user_id, payload):
    app.cable.broadcast(f"inbox_{user_id}", payload)

The controller saves the message and redirects as usual, and the broadcast updates every open page. Background Tasks covers running the worker that the second example needs.

It doesn't matter which process the broadcast comes from. Only one process serves the WebSockets; a broadcast made in any other one - a task worker, a shell, the extra web processes of PROCESSES - is sent to it for you (see The cable server).

When one request makes several broadcasts, wrap them in app.cable.batch(). From a process that has to send them to the cable, they then travel together, in one request instead of one each:

with self.app.cable.batch():
    self.app.cable.broadcast(f"room_{room.id}", html)
    for user_id in member_ids:
        self.app.cable.broadcast(f"unreads_{user_id}", {"room_id": room.id})

7.4 Closing a user's connections

A channel authorizes when it subscribes. If you take that access away later - remove someone from a room, ban them, sign them out - their open subscriptions keep receiving until they disconnect. app.cable.disconnect(user_id=...) closes every connection of that user, from any process. Their pages reconnect and subscribe again, and the channels that no longer authorize them reject the subscription:

membership.delete_instance()
self.app.cable.disconnect(user_id=membership.user_id)

8. Who is here: presence

A channel can list its connection among those present in a stream, with track(). The server keeps the list, tells the stream's subscribers who joins and leaves, and gives every new subscriber the list as it stands:

@router.channel()
class RoomChannel(AppChannel):
    def subscribed(self):
        if not self.authenticated:
            self.reject()
            return
        stream = f"room_{self.params['room_id']}"
        self.stream_from(stream)
        self.track(stream, {"name": current.user.name, "avatar": current.user.avatar_url})

    def away(self, data):
        self.update_presence({"name": current.user.name, "status": "away"})

cable.subscribe("RoomChannel", { room_id: 42 }, {
  presence(users, change) {
    renderWhoIsHere(Object.values(users))  // {key: data}, as of now
    if (change?.event === "join") toast(`${change.data.name} is here`)
  },
})

The presence callback runs with the whole list when the subscription is confirmed (change is null) and each time it changes; change is {event, userId, data}, with event one of join, leave or update. On the server, app.cable.presence(stream) returns {key: {"data": ..., "connections": n}} and app.cable.presence_stats(stream) the counts, from any process that serves the WebSockets.

What the list keys on:

• A logged-in user is listed by their id. Three tabs are one entry: join goes out with the first, leave with the last. That is also what unsubscribed() can't give you, because it runs once per tab.
• An anonymous connection gets a random key, so a visitor with three tabs counts three times.
• track(stream, data, key=...) lists the connection under a key of your choosing, for example a bot's name. The key is decided in subscribed(), on the server. Never take it from params: a client could pose as anyone. One key per connection; a second track() with another is an error.

A connection leaves a stream's list with untrack(stream), with stop_stream_from(), when the client unsubscribes, and when the connection closes, including when the browser is killed: the server notices the connection is gone and removes the entry. update_presence(data) changes the data in every stream the connection is present in. track() needs stream_from() first.

presence=False in CABLE turns it off (track() then fails). The data of a user may be 4 KB at most as JSON (presence_max_data_size), and presence_max_members caps the users of a stream (0, the default, is no limit); past either, track() and update_presence() raise ValueError. With RedisCable the list is one for all the machines (see Several machines).

9. The client: cable.js

cable.js is an ES module that owns the connection, your subscriptions, and reconnection. It is in the import map as "cable":

import { cable } from "cable"

cable.connect()

const chat = cable.subscribe("ChatChannel", { room: "general" }, {
  connected()    { console.log("subscribed") },
  disconnected() { console.log("connection lost") },
  rejected()     { console.log("subscription denied") },
  received(data) { addMessageToDOM(data) },
})

chat.perform("speak", { message: "hello" })  // call an action; a promise with its reply
chat.send({ message: "hello" })   // shorthand for perform("receive", data)
chat.unsubscribe()                // leave this channel
cable.disconnect()                // close the socket, stop reconnecting

cable.connect() opens the WebSocket. Calling it again while it is open, or opening, does nothing, so every script of the page can call it. With no argument, it connects to /cable on the current host (ws:// on http, wss:// on https); in development, to the port the page announces (see The cable server). If you change CABLE_PATH, pass the full URL to connect().

cable.subscribe(channel, params, callbacks) returns a subscription. You can subscribe before the connection is open: the subscription is sent as soon as it is. If you pass only two arguments and the second has callbacks in it, it is taken as the callbacks, with empty params.

The four callbacks are optional:

When the connection drops, cable.js reconnects on its own, subscribes everything again, and gets the broadcasts it missed (see Missed broadcasts), so a short network problem is invisible to your code. A perform() made meanwhile waits, and is sent once the connection and its subscriptions are back; its promise still rejects with {reason: "timeout"} if that takes longer than its timeout, and with {reason: "offline"} if more than 100 calls pile up. It waits 1 second before the first attempt and twice as long before each next one, up to 30 seconds, and it never gives up. The server pings every connection every CABLE_PING_INTERVAL seconds; a connection that goes silent for three intervals (and at least 10 seconds) is treated as dead, closed, and opened again. This catches a laptop that went to sleep or a proxy that dropped the connection without telling anyone. cable.disconnect() is the only thing that stops the reconnecting.

All subscriptions share the one connection, so holding several is normal and cheap:

const general = cable.subscribe(
    "ChatChannel", { room: "general" }, { received: render })
const random  = cable.subscribe(
    "ChatChannel", { room: "random" },  { received: render })
const inbox   = cable.subscribe(
    "InboxChannel", { received: showToast })

9.1 Where the JavaScript goes

New apps load assets/js/application.js as a module from the layout and navigate with Turbo Drive, which swaps the <body> on each click instead of reloading. Two consequences for subscriptions:

• cable.connect() and anything every page needs belong in application.js. The module runs once per real page load, and the connection survives Turbo navigations, so a tab opens one WebSocket and keeps it.
• A subscription for one page must end when that page goes away. A cable.subscribe() at the top of application.js would subscribe once and never let go; one inside a turbo:load handler would subscribe again on every visit, piling up. Tie it to an element instead, so it unsubscribes when the element leaves the DOM. For HTML broadcasts, <turbo-stream-channel> does exactly that (see Broadcasting HTML). For JSON, a Stimulus controller, which new apps also include, has the same two moments:

import { Controller } from "@hotwired/stimulus"
import { cable } from "cable"

class RoomController extends Controller {
  static values = { id: Number }

  connect() {
    this.subscription = cable.subscribe("RoomChannel", { room_id: this.idValue }, {
      received: (data) => this.render(data),
    })
  }

  disconnect() {
    this.subscription.unsubscribe()
  }

  render(data) { ... }
}

window.Stimulus.register("room", RoomController)

{#js "js/room-controller.js" #}

<div data-controller="room" data-room-id-value="{{ room.id }}">...</div>

The view loads the controller with the {#js #} tag, as the auth addon's password.jx does, and the element's data-controller attribute starts it. When Turbo replaces the body, disconnect() runs and the subscription ends with the page.

9.2 Missed broadcasts (recovery)

The server keeps the last broadcasts of each stream: 128 by default, for 5 minutes after the last one. When a connection comes back, cable.js tells the server where it was in each stream, and the server sends what was broadcast since. They arrive through received() like any other, in order, and nothing arrives twice. A connection that falls too far behind is closed (see Slow clients), and gets what it missed when it reconnects, the same way.

connected() is where the page learns whether it is up to date. It gets an object:

cable.subscribe("ChatChannel", { room: "general" }, {
  connected({ reconnected, recovered }) {
    if (reconnected && recovered === false) loadMessages()  // too much was missed
  },
  received: render,
})

With recovered === false, load the state again from the server, as the page did when it first rendered. The <turbo-stream-channel> element does this by dispatching an event (see Broadcasting HTML).

What recovery does not cover: a stream that had no broadcasts before the connection dropped has no position to recover from, so what it broadcast meanwhile is lost (recovered is null for it); and send() messages to one subscription are not kept, only broadcasts. On the server, recovery=False in CABLE turns it off, and recovery_buffer_size, recovery_ttl and recovery_memory_budget size the buffers (see The cable server).

10. Broadcasting HTML (Turbo Streams)

The broadcasts so far send JSON, and the received() callback rebuilds the DOM by hand - writing again, in JavaScript, markup you already have as a Jx component.

Proper includes Turbo, so instead you can broadcast the rendered component wrapped in a <turbo-stream> and let Turbo apply it. A live list then needs no JavaScript of your own.

A <turbo-stream> is HTML that names a DOM operation and a target:

<turbo-stream action="append" target="messages">
  <template><li>Ana: hello</li></template>
</turbo-stream>

The turbo_stream builder has a method for each operation. Each one renders a Jx component, or takes ready-made HTML:

from proper import turbo_stream

turbo_stream.append(
    "messages",       # the id of the element to update
    "message.jx",     # the component to render
    message=message,  # arguments for the component
)
# -> <turbo-stream action="append" target="messages">
#      <template>...</template>
#    </turbo-stream>

turbo_stream.update("unread_count", html="3")
turbo_stream.remove(message)  # a model: its dom_id is the target
turbo_stream.remove(dom_id(message, "list"))
turbo_stream.replace(targets=".draft", html="")  # every element matching a CSS selector

The first argument is the id of the element to act on, or a model instance, whose dom_id is used. targets= takes a CSS selector instead, to act on every matching element. The method name is the operation:

Broadcast it like any other data:

from proper import current, turbo_stream

...

    def create(self):
        room_id = self.params["room_id"]
        message = Message.create(
            room_id=room_id,
            text=self.params["text"],
            author=current.user,
        )

        self.app.cable.broadcast(
            f"chat_{room_id}",
            turbo_stream.append("messages", "message.jx", message=message),
        )
        self.response.redirect_to("Message.index", room_id=room_id)

To subscribe a page, you don't need to write JavaScript either. Use the <turbo-stream-channel> element, which cable.js defines:

{#import "message.jx" as Message #}

<turbo-stream-channel channel="ChatChannel" params='{"room_id": 42}'>
</turbo-stream-channel>

...

<ul id="messages">
  {% for message in messages %}
    <Message message={{ message }} />
  {% endfor %}
</ul>

The element connects, subscribes to the channel with those params, and hands every message to Turbo, which appends the new <li>. When the element leaves the page, it unsubscribes. The same Message component renders the initial list and every live update, so the markup is defined in one place.

To subscribe from JavaScript instead, use streamFrom:

import { streamFrom } from "cable"

const subscription = streamFrom("ChatChannel", { room_id: 42 })

Authorization doesn't change: the element sends params, never a stream name, and the channel's subscribed() still decides with reject() and picks the stream on the server (see Authorizing versus authenticating).

After a reconnection, the <turbo-stream> fragments the page missed are applied as they arrive. When they can't be recovered (see Missed broadcasts), the element dispatches a turbo-stream-channel:gap event, which bubbles, so the page can load the content again:

document.addEventListener("turbo-stream-channel:gap", (event) => {
  Turbo.visit(location.href, { action: "replace" })  // or reload a frame
})

streamFrom(channel, params, { onGap }) takes the same as a function.

11. The cable server: Cable

The channels addon configures this cable:

CABLE_PATH = "/cable"
CABLE_PORT = int(os.getenv("CABLE_PORT", int(os.getenv("PORT", 2300)) + 1))
CABLE: dict = {"type": "proper.channels.Cable"}

Cable serves the WebSockets with Bearings, the same server that runs the app, inside the web process. Your channels run in Python, in its worker threads. Underneath, a stream is a Bearings topic, and a broadcast() gives Bearings one encoded frame, which it writes to every subscriber without going back to Python. With thousands of clients on one stream, this is what keeps up.

How it runs:

• proper run starts the cable in its web process, before the first request, and stops it with the server. The cable listens on its own port, CABLE_PORT. Under another server, call app.cable.start_server() and app.cable.stop_server() yourself. start_server() returns once the cable listens, and raises OSError, with the reason, when the port can't be listened on (it is taken, or the cable's host is not an address of this machine).
• Only that process serves the WebSockets. Every other process that loads the app - the extra copies of PROCESSES, a task worker, a shell - sends its broadcasts and disconnect()s to it, signed with the app's secret keys, as a POST to CABLE_PATH on 127.0.0.1, port CABLE_PORT + 1. If the cable is down, the broadcast is lost and a warning is logged; the request that made it still finishes normally. One whose answer takes more than 2 seconds (the cable answers once it has delivered it) is sent again, and the cable ignores the copy.
• In production, your reverse proxy routes CABLE_PATH to CABLE_PORT. Deployment shows the nginx block.
• In development there is no proxy. When DEBUG is on, render_importmap() adds a <meta name="cable-port"> tag to the page, and cable.js connects to that port on the same host name.

The settings, in config/:

And the options of Cable, in CABLE:

CABLE = {
    "type": "proper.channels.Cable",
    "loops": 8,
    "workers": 8,
}

loops decides how many broadcasts per second the cable delivers. On a 16-core machine, one stream with 1000 clients got about 5 million messages a second with one loop, 14 million with the default 4, and 25 million with 16; each costs about the same CPU. The cable shares the machine with the threads that run your app, so raise it when the WebSockets are the busy part, and lower it on a small server.

11.1 How the cable is doing

app.cable.health() returns a dict for a health endpoint or a log line: serving (whether this process serves the WebSockets), connections, users (with a logged-in user), streams (subscribed here), and under server the server's counters: connections, and how many streams have a recovery buffer and a presence list. In a process that serves none, {"serving": False, "connections": 0, "streams": 0}.

@router.get("/up")
def up(self):
    return self.render_json(self.app.cable.health())

For graphs and alerts over time, the cable also exports Prometheus metrics. The process that serves the WebSockets answers GET /metrics on the loopback port the other processes forward to, 127.0.0.1:CABLE_PORT + 1: proper_cable_connections, proper_cable_streams and proper_cable_users. Point Prometheus at that port on each machine. It is a loopback port, so the metrics are never exposed to the outside. app.cable.metrics() returns the same text, empty where nothing is served.

scrape_configs:
  - job_name: proper-cable
    static_configs:
      - targets: ["10.0.0.1:2302", "10.0.0.2:2302"]

11.2 Slow clients

A client that stops reading - a frozen tab, a very bad network - would make messages pile up in memory. So each connection has a limit, CABLE_MAX_PENDING_BYTES (4 MB): a client that falls that far behind is closed. cable.js reconnects and subscribes again, and gets what it missed. A client that reads slowly but keeps up stays.

12. Several machines: RedisCable

Cable serves the WebSockets of one machine. To run the app on several, behind a load balancer, use RedisCable: the same cable, with Redis carrying the broadcasts between machines. Your channels and your code don't change, only the config:

CABLE = {
    "type": "proper.channels.RedisCable",
    "url": os.getenv("REDIS_URL", "redis://localhost:6379/0"),
    "prefix": "myapp:cable:",
}

It needs the redis package (uv add redis).

Each machine's web process serves its own WebSockets, as with Cable. A broadcast() goes to the subscribers of the process that makes it, if it serves any, and to Redis, from where the web processes of the other machines deliver it to theirs. A process without WebSockets - a task worker, a shell - only publishes to Redis, so nothing is sent over CABLE_PORT + 1. disconnect() reaches every machine the same way.

It takes every option of Cable as well.

If the connection to Redis drops, the cable reconnects on its own, waiting longer each time, up to 30 seconds. Meanwhile, each machine's clients still get the broadcasts made on that machine; what other machines publish in that time is lost, and a broadcast that can't reach Redis is lost with a warning.

Each machine keeps its own recovery buffers, with their own positions. A client that reconnects to the same machine gets what it missed; one the load balancer sends to another machine gets recovered: false, and should load the state again. Pin a client to a machine (sticky sessions on CABLE_PATH) if that matters.

Presence is one list for all the machines, kept in Redis: presence() shows who is in a stream anywhere, and a user joins with their first connection on any machine and leaves with their last. A machine that dies is noticed within 15 seconds, and its users leave. Without Redis, each machine lists only its own.

Keep in mind that RedisCable shares broadcasts between machines, not state. Redis pub/sub carries messages; it doesn't store them. A list of who is online across machines, or the last value for a client that subscribes late, is something you keep yourself.

13. Checking where a WebSocket comes from

Browsers send a site's cookies with any WebSocket to it, even one opened by a page of another site. Without a check, a malicious page could open a WebSocket to your app with your user's session. So the cable checks the Origin header of the handshake, and accepts:

• A handshake with no Origin (it doesn't come from a browser).
• One from the same host the WebSocket was opened to, or from the app's HOST.
• In DEBUG, one from localhost or 127.0.0.1 on the app's PORT.
• One listed in CABLE_ALLOWED_ORIGINS.

Any other is refused with a 403, and so is an Origin that is not a valid URL (http://[abc]). To allow another site, list it:

CABLE_ALLOWED_ORIGINS = ["https://admin.example.com"]

14. Without channels

An app without the channels addon has CABLE = {}: a cable that serves no WebSockets, so broadcast() reaches no one (it is logged at the debug level) and doesn't fail. Setting CABLE_PORT with an empty CABLE is a ConfigError when the app starts.

15. When nothing arrives

A broadcast to a stream nobody listens to is not an error: it reaches no one, and nothing is logged. So when a page doesn't update, there is no message to read, and it helps to check in this order:

1. Is the page connected? In the browser's developer tools, the Network panel has a WS filter. The cable connection should be there with status 101, and its Messages tab shows every frame in both directions: your subscribe command, the server's confirm_subscription (or reject_subscription), the pings, and each broadcast. In development, the connection goes to CABLE_PORT (2301), announced by a <meta name="cable-port"> tag in the page; in production, to CABLE_PATH through your proxy (see Deployment).
2. Was the subscription confirmed? A reject_subscription means subscribed() called reject(), raised (the server log has the traceback), or no channel has that name: the channel module isn't imported in channels/__init__.py, or the name differs from the class name (@router.channel("chat") registers chat, not ChatChannel). The confirmation lists the subscription's streams: if the one you broadcast to isn't there, subscribed() didn't stream_from() it.
3. Do the names match? The stream in stream_from() and the one in broadcast() must be the same string. f"chat_{room.id}" and f"chat_{room_id}" look alike and differ when one is an int and the other a string from the URL. In a shell, app.cable.streams shows the streams with subscribers right now, and how many.
4. Is the broadcast reaching the cable? From a task worker or a shell, the broadcast is forwarded to the web process; a warning in the log says when it can't be reached. app.cable.health() says whether a process serves the WebSockets and how many connections it has.
5. Is it the data? received() gets exactly what you broadcast, after a round trip through JSON: a date becomes a string, a model has to be turned into a dict first. The Messages tab shows the frame as it arrived.

16. Testing channels

You don't need a running server to test a channel. The test client opens a WebSocket session served from memory: it subscribes, calls actions, and disconnects, and you check the frames that come back.

These tests are async, so they need pytest-asyncio:

$ uv add --dev pytest-asyncio

import pytest


@pytest.mark.asyncio
async def test_speak_broadcasts_to_the_room(client):
    ws = client.websocket()
    task = await ws.connect()

    confirm = await ws.subscribe("ChatChannel", room="general")
    assert confirm["type"] == "confirm_subscription"

    await ws.send_action(
        "ChatChannel",
        "speak",
        {"message": "hi"},
        room="general",
    )
    msg = await ws.receive()
    assert msg["type"] == "broadcast"
    assert msg["data"] == {"message": "hi"}

    await ws.close()
    await task

The session's methods:

subscribe() returns the first frame the app sends. If subscribed() calls send(), that message comes first, and the confirmation is the next receive().

Several sessions in one test share the same cable, so you can open two and check that a broadcast reaches both. A broadcast from your test code, client.app.cable.broadcast(...), reaches them too.

To test an authenticated channel, sign a user in first: the handshake carries the same session cookie an HTTP request would.

@pytest.mark.asyncio
async def test_anonymous_users_are_rejected(client):
    ws = client.websocket()
    await ws.connect()
    reply = await ws.subscribe("InboxChannel")
    assert reply["type"] == "reject_subscription"
    await ws.close()


@pytest.mark.asyncio
async def test_users_get_their_inbox(client, session):
    client.sign_in(session)
    ws = client.websocket()
    task = await ws.connect()
    reply = await ws.subscribe("InboxChannel")
    assert reply["type"] == "confirm_subscription"
    assert reply["streams"] == [f"inbox_{session.user_id}"]
    await ws.close()
    await task

The session runs the app's cable from memory (app.cable.serve_in_memory()): no port, no threads, and each message is handled right away. With an app whose cable serves no WebSockets (CABLE = {}), connect() raises a RuntimeError. The Testing guide covers the TestClient and sign_in().

17. The wire protocol

cable.js speaks this so you don't have to, but the frames are useful to know for debugging, or for writing a client in another language. Every message is a JSON object.

The client sends three commands - subscribe, message (call an action), and unsubscribe:

{ "command": "subscribe", "channel": "ChatChannel",
    "params": {"room": "general"},
    "positions": {"chat_general": [{"e": "0000abcd", "o": 41}]} }

{ "command": "message", "channel": "ChatChannel",
    "params": {"room": "general"}, "action": "speak",
    "data": {"message": "hi"}, "id": 7 }

{ "command": "unsubscribe", "channel": "ChatChannel",
    "params": {"room": "general"} }

The id of a message is optional: with one, the server answers with a reply.

The server sends back confirm_subscription, reject_subscription, message (from send()), broadcast (from broadcast()), reply and error, plus the cable's ping:

{ "type": "confirm_subscription", "channel": "ChatChannel",
    "params": {"room": "general"}, "streams": ["chat_general"],
    "positions": {"chat_general": {"e": "0000abcd", "o": 42}},
    "recovered": true, "presence": {"chat_general": {"7": {"name": "Ana"}}} }

{ "type": "reject_subscription", "channel": "ChatChannel",
    "params": {"room": "general"} }

{ "type": "message", "channel": "ChatChannel",
    "params": {"room": "general"}, "data": {"message": "hi"} }

{ "tp": "chat_general", "e": "0000abcd", "o": 42,
    "c": "P", "type": "broadcast", "stream": "chat_general",
    "data": {"message": "hi"} }

{ "type": "reply", "id": 7, "channel": "ChatChannel",
    "params": {"room": "general"}, "status": "ok", "data": {"id": 42} }

{ "type": "reply", "id": 7, "channel": "ChatChannel",
    "params": {"room": "general"}, "status": "error",
    "data": {"reason": "too_long", "max": 500} }

{ "type": "error", "reason": "not_subscribed" }

{ "c": "WSE", "t": "ping", "p": {} }

{ "c": "WSE", "t": "presence_join", "p": {"topic": "chat_general",
    "user_id": "7", "data": {"name": "Ana"}} }

A broadcast is one frame, the same for every subscriber, so it can't carry each subscription's channel and params. It names the stream instead, and confirm_subscription lists the streams of the subscription, so the client delivers a broadcast to every subscription that streams from it. Ignore the c field.

The tp, e and o fields are the recovery stamp: the stream, the epoch of its buffer (eight hex digits; it changes when the server restarts) and the offset of the broadcast in it. A subscribe may carry positions, the last o the client saw per stream and epoch (a list per stream, because on several machines each stamps its broadcasts with its own epoch; one {"e", "o"} is accepted too); the server sends again the broadcasts after them, for the streams the channel streams from, and they may arrive before the confirmation. The confirmation has the current positions of the streams (null for one without broadcasts yet) and recovered: true when everything asked for was sent again, false when some couldn't be, null when nothing was asked. Sending subscribe again for an existing subscription, with positions, also recovers. Without recovery (recovery=False), broadcasts carry no stamp and recovered is always null.

Subscribing again to a subscription that already exists only sends its confirmation again. A reject_subscription has "reason": "unknown_channel" when no channel has that name, and "reason": "error" when subscribed() raised an exception; a subscription your own reject() turned away has no reason.

A reply answers the message with the same id: status: "ok" with what the action returned as data (null if nothing), or status: "error" with a reason in data: the one of an ActionError, along with its other data; error when the action raised anything else; or not_subscribed, invalid_action or unknown_action when there was nothing to run.

A message without an id gets no reply. What would have been an error reply is an error frame instead, with the reason (and the data of an ActionError) at the top level; an action that raises anything else sends nothing. An error also reports a command the server couldn't read: invalid_json, invalid_message (JSON that is not an object), or unknown_command.

The confirmation's presence lists who is in each stream the channel tracks, {key: data}. The presence_join, presence_leave and presence_update frames go to every subscriber of the stream (topic), when a user's first connection is tracked, their last one leaves, or their data changes; they may arrive before the confirmation of the subscription that caused them.

The ping is the cable's, every CABLE_PING_INTERVAL seconds. The client answers it with {"c": "WSE", "t": "PONG", "p": {}}; a connection that answers none for idle_timeout seconds is closed. cable.js also uses it to tell a dead connection from a quiet one: three intervals of silence (and at least 10 seconds), and it reconnects.

18. A full example

A room chat: an authenticated channel, a controller that saves a message and broadcasts it, and the page that ties them together.

from proper import current

from ..models import Room
from ..router import router
from .app_channel import AppChannel


@router.channel()
class ChatChannel(AppChannel):
    def subscribed(self):
        room = Room.get_or_none(Room.id == self.params["room_id"])
        if (
            room is None
            or not self.authenticated
            or not room.has_member(current.user)
        ):
            self.reject()
            return
        self.stream_from(f"chat_{room.id}")

from proper import current, turbo_stream

from ..models import Message
from ..router import router
from .app_controller import AppController


@router.resource("rooms/:room_id/messages")
class MessageController(AppController):
    def create(self):
        room_id = self.params["room_id"]
        message = Message.create(
            room_id=room_id,
            author=current.user,
            text=self.params["text"],
        )
        self.app.cable.broadcast(
            f"chat_{room_id}",
            turbo_stream.append("messages", "message.jx", message=message),
        )
        self.response.redirect_to("Message.index", room_id=room_id)

{#def room, messages #}
{#import "message.jx" as Message #}

<turbo-stream-channel channel="ChatChannel" params='{"room_id": {{ room.id }}}'>
</turbo-stream-channel>

<ul id="messages">
  {% for message in messages %}
    <Message message={{ message }} />
  {% endfor %}
</ul>

<form method="post" action="{{ url_for('Message.create', room_id=room.id) }}">
  <input name="text" autocomplete="off">
  <button>Send</button>
</form>

The channel guards the room with the server-verified user. The controller saves the message and broadcasts it, rendered with the same component as the list. Every open page of the room, including the sender's, gets the new message, and there is no JavaScript of your own.

19. Where this could grow

Channels covers the core - one shared connection, authenticated subscriptions, streams, broadcasting, a reconnecting client, recovery of missed broadcasts, replies to actions, and Turbo Streams. Some things that other real-time stacks offer are not here yet. None of them block you - there are workarounds - but they are where the framework will likely grow.

• Stream names from models. broadcast_to(record, data) and stream_for(record), which would build the stream name from a model, so a typo can't break it silently.
• Per-subscription timers. A periodically(...) hook for server-driven updates - counters, clocks, dashboards - that lives as long as the subscription. For now, a periodic background task that broadcasts to a stream covers most of this.

20. What's next

Channels touches several other parts of Proper:

• Authentication - the session and signed cookie that AppChannel uses to put current.user on a connection.
• Turbo - Turbo Frames and Streams, also as responses to forms.
• Background Tasks - the worker behind broadcasting from a task, plus scheduling and retries.
• Jx Components - the components you render into a <turbo-stream> to broadcast.
• Deployment - processes, and the cable port behind nginx.