Channels

Most of the time, Pulse's reactive state system handles everything you need. You update state on the server, and the UI updates automatically. But what happens when you need the server to push updates to the client without waiting for user interaction?

Consider these scenarios:

A chat application where messages from other users should appear instantly
A notification system that alerts users when something happens elsewhere in the system
A live dashboard that updates when new data arrives on the server
A collaborative document where multiple users see each other's changes in real-time

These all share a common need: server-initiated communication. The server knows something happened and needs to tell the client right now, not when the client next makes a request. That's what channels are for.

Why channels instead of regular state?

You might wonder: can't I just update state and let Pulse re-render?

State works great when the user triggers the change. User clicks a button, handler runs, state updates, UI re-renders. But state updates only reach the user who triggered them. If Alice sends a message, Bob won't see it unless he refreshes or does something to trigger his own re-render.

Channels solve this by creating a direct communication line between server and client:

Pattern	Best for	How it works
State	Single-user interactions	User action triggers state change and re-render
Channels	Multi-user or server-initiated updates	Server pushes data to connected clients in real-time

Think of state as a one-way mirror: the user sees their own reflection. Channels are more like a window: everyone can see through to the other side.

A real-world scenario: live notifications

Let's make this concrete. Imagine you're building a project management app. When a teammate assigns a task to you, you want to see a notification immediately, not the next time you click something.

Here's how you'd build this with channels:

import pulse as ps

class NotificationState(ps.State):
    notifications: list[dict] = []

    def add(self, notification: dict):
        self.notifications = [notification] + self.notifications

    def dismiss(self, index: int):
        self.notifications = [
            n for i, n in enumerate(self.notifications) if i != index
        ]


@ps.component
def NotificationCenter():
    with ps.init():
        channel = ps.channel("notifications")
        state = NotificationState()

    # When the server sends a notification, add it to state
    channel.on("new_notification", lambda payload: state.add(payload))

    return ps.div(className="fixed top-4 right-4 space-y-2")[
        ps.For(
            state.notifications,
            lambda notif, i: ps.div(
                className="bg-blue-100 border border-blue-300 rounded p-3 shadow"
            )[
                ps.p(notif["message"]),
                ps.button(
                    "Dismiss",
                    onClick=lambda: state.dismiss(i),
                    className="text-sm text-blue-600 underline",
                ),
            ],
        ),
    ]

Now, anywhere in your backend code, you can send a notification:

# In some other part of your app, maybe when a task is assigned:
def assign_task(task_id: str, assignee_user_id: str):
    # ... save to database ...

    # Send real-time notification to the assignee
    # (assuming you have a reference to their notification channel)
    channel.emit("new_notification", {
        "message": f"You've been assigned task #{task_id}",
        "type": "task_assignment",
    })

The notification appears instantly for the user, without them clicking anything.

Creating a channel

Create a channel inside a component using ps.channel(). Each channel is automatically scoped to the current user session and route.

import pulse as ps

@ps.component
def ChatRoom():
    with ps.init():
        chat = ps.channel("chat")  # The name is optional, helpful for debugging

    # ... rest of component

The string argument ("chat") is an identifier that helps with debugging. If you don't provide one, Pulse generates a UUID automatically. The channel is cleaned up automatically when the component unmounts or the user navigates away.

Handling events from the client

Use channel.on() to register handlers for events sent from the browser. This is how you receive messages from client-side JavaScript code.

@ps.component
def ChatRoom():
    with ps.init():
        chat = ps.channel("chat")
        messages = ChatState()

    # Handle incoming messages from the client
    chat.on("send_message", lambda payload: messages.add(payload["text"]))

    # Handlers can be async for operations that need to await
    async def handle_typing(payload):
        await messages.set_typing(payload["user"])

    chat.on("typing", handle_typing)

Each handler receives a payload dictionary containing whatever data the client sent. You can register multiple handlers for the same event if needed.

Sending events to the client

To push data from server to client, use channel.emit(). This is a fire-and-forget operation: you send the data and move on.

def notify_users(chat: ps.Channel, message: str):
    chat.emit("new_message", {"text": message, "timestamp": time.time()})

Sometimes you need the client to respond with data. Maybe you want to know the user's current scroll position, or you need to read a value from local storage. For this request-response pattern, use channel.request():

async def get_client_info(chat: ps.Channel):
    # Request data from the client and wait for the response
    response = await chat.request("get_info", timeout=5.0)
    return response

The timeout parameter prevents your code from waiting forever if the client doesn't respond.

Using channels on the client

When you need custom client-side logic to interact with a channel, you'll write JavaScript code that runs in the browser. Pulse provides the @ps.javascript decorator for embedding JavaScript in your Python code.

The usePulseChannel hook connects your JavaScript to a channel:

from pulse.js.pulse import usePulseChannel

@ps.javascript(jsx=True)
def ChatClient(*, channelId: str):
    bridge = usePulseChannel(channelId)

    # Listen for events from the server
    def on_new_message(payload):
        # This runs in the browser when the server emits "new_message"
        console.log("Got message:", payload)

    useEffect(lambda: bridge.on("new_message", on_new_message), [bridge])

    # Send events to the server
    def send_message(text: str):
        bridge.emit("send_message", {"text": text})

    # Request data from the server and wait for response
    async def request_history():
        history = await bridge.request("get_history")
        return history

A note on @ps.javascript: This decorator lets you write JavaScript logic that runs in the browser. The code inside looks like Python but transpiles to JavaScript. This is useful when you need browser APIs (like localStorage, window, or DOM manipulation) or want to optimize performance by keeping some logic client-side.

The usePulseChannel hook returns a bridge object with three methods:

on(event, handler) - Listen for events from the server
emit(event, payload) - Send events to the server (fire-and-forget)
request(event, payload) - Send a request and await the server's response

Channel lifecycle

Channels are tied to the route where they're created. When the user navigates to a different route, the channel closes automatically and resources are cleaned up. This prevents memory leaks and stale connections.

If you need a channel that persists across route changes (like a global notification system), create it at a higher level in your component tree, such as in a layout component.

You can also close a channel manually when you're done with it:

chat.close()

Error handling

Network connections aren't always reliable. When using channel.request(), the client might be slow or unresponsive. Always handle timeouts:

from pulse import ChannelTimeout

async def safe_request(chat: ps.Channel):
    try:
        response = await chat.request("get_data", timeout=5.0)
        return response
    except ChannelTimeout:
        # Client didn't respond in time
        return None

You can also catch ChannelClosed if you need to handle the case where someone tries to use a channel that's already been closed.

When to use channels

Use channels when you need:

Real-time updates from the server - notifications, live data feeds, server-sent alerts
Multi-user communication - chat, collaborative editing, presence indicators
Request-response with the browser - getting client-side data like viewport size, local storage, or user preferences

Stick with regular state when:

Updates are triggered by the current user's actions
You don't need other users to see changes in real-time
The server doesn't need to push unsolicited updates