How to implement server push ?

[PasqualePuzio]

We are trying to figure what's the best way to implement server pushes, meaning that we want our application to notify the client whenever there's an update.

In other words, we want our application to send an updated state to the client whenever there's new data to be handled.

In more practical terms, we would like to open a websocket on the server (I guess it isn't possible to do that on the client for now) and keep listening for new messages. Is this the recommended approach?

[masenf]

A reflex app already maintains a bidirectional websocket connection between the frontend and backend. The best mechanism to notify the client depends on the nature of the "push". The simplest way to achieve this is using @rx.background on an event handler that polls for some condition and updates the user's state in response. Updating the state in a background task is a pushed update, by definition.

Let's say the client wants a push whenever the github zen API changes:

import asyncio
import httpx

import reflex as rx


ZEN_API = "https://api.github.com/zen"
LATEST_ZEN = "Life is short (so use Python)"
UPDATE_TASK = None


async def periodic_update_github_zen():
    global LATEST_ZEN

    async with httpx.AsyncClient() as client:
        while True:
            resp = await client.get(ZEN_API)
            resp.raise_for_status()
            LATEST_ZEN = resp.text
            print(f"Updated zen on backend to: {LATEST_ZEN}")
            await asyncio.sleep(10)  # don't overwhelm the API


class State(rx.State):
    zen: str = LATEST_ZEN
    _poll_updates: bool = True
    _task_running: bool = False

    @rx.background
    async def push_zen_updates(self):
        if self._task_running:
            return

        async with self:
            self._task_running = True

        try:
            while self._poll_updates:
                async with self:
                    if self.zen != LATEST_ZEN:
                        self.zen = LATEST_ZEN
                await asyncio.sleep(5)
        finally:
            async with self:
                self._task_running = False


@rx.page(on_load=State.push_zen_updates)
def index() -> rx.Component:
    return rx.vstack(
        rx.heading("Your Momentary Zen", font_size="2em"),
        rx.text(State.zen),
        spacing="1.5em",
    )


app = rx.App()


@app.api.on_event("startup")
def start_updater():
    global UPDATE_TASK
    UPDATE_TASK = asyncio.create_task(periodic_update_github_zen())


@app.api.on_event("shutdown")
def stop_updater():
    UPDATE_TASK.cancel()


app.compile()

This example uses a global task to poll the github zen API, and then each tab uses its own background task to update itself when the global zen value changes.

This simulates individual users polling a database or something for an update, then applying the update to their state instance, which automatically sends a push to the client.

[thevicchij]

Hi @masenf,

Thank you, this was very helpful!

I'm trying to setup a polling loop that should be active as long as the user is on a given page. However, I've not found a reliable way to stop it. The on_unmount hook works as long as the user navigates within the app, but is not called if e.g. the user closes their tab.
What would be a good approach to stop polling in this scenario?

[masenf]

@thevicchij another option, if you want a periodic callback is to leverage the rx.moment component.

Normally this component is used to format datetime, but it can be set to automatically update itself, and when it updates it can fire a callback on_change. So you can hide the date display, and use this feature to get a callback on your backend that is initiated by the client without any long running task on your server.

import reflex as rx


class State(rx.State):
    auto_count: int = 0

    def tick(self, date):
        """Increment the auto_count."""
        self.auto_count += 1


def index() -> rx.Component:
    return rx.vstack(
        rx.heading(rx.moment(interval=1000, on_change=State.tick), display="none"),
        rx.text(State.auto_count),
    )


app = rx.App()
app.add_page(index)

[giovanni-steerable-co]

Sorry for the late reply, I had to work on something else..

It works great, thank you so much for the hint!