Push to browsers¶

Push is an ASGI app that streams events to browsers as Server-Sent Events. A browser that reconnects resumes where it left off, within the events' retention of an hour.

Serving a stream¶

The service that serves the stream declares events = Events(), even if it publishes nothing, and passes its bus to Push with the event types to send. With Litestar, mount it as an ASGI handler, with its arguments annotated:

from litestar import Litestar, asgi
from litestar.types import Receive, Scope, Send

from myxo.events import Events
from myxo.push import Push


class Web(Service):
    name = "web"
    stage = 2
    Config = WebConfig
    events = Events()

    async def run(self, ready: Callable[[], None]) -> None:
        push = Push(self.events, (Noted,))

        @asgi("/stream", is_mount=True)
        async def stream(scope: Scope, receive: Receive, send: Send) -> None:
            await push(scope, receive, send)

        app = Litestar([stream])
        address = ("127.0.0.1", self.config.port)
        await serve_asgi(app, address, ready, self.runtime.stopping.set)

On Kubernetes, read the address from the Config, as that guide shows. Like every framework service, Events() needs the layout's top-level database, which must be the one the publishers write to. The publisher may run in another process: the events go through the database.

What travels¶

Each event arrives with its position as its id, its type's name, and its JSON:

$ curl -sN localhost:8000/stream
id: 1
event: Noted
data: {"note_id":1}

id: 2
event: Noted
data: {"note_id":2}

With Last-Event-ID, a client resumes after that position:

$ curl -sN -H 'Last-Event-ID: 1' localhost:8000/stream
id: 2
event: Noted
data: {"note_id":2}

A browser's EventSource does this for you. Listen by the event's type:

const stream = new EventSource("/stream");
stream.addEventListener("Noted", (e) => console.log(JSON.parse(e.data)));

Behaviour¶

• Every 15 s without an event, the stream sends a comment, which keeps proxies from closing it. Push(..., keepalive=30.0) changes the delay.
• The stream ends when the client leaves, or when the process begins to stop. That is why a web service passes self.runtime.stopping.set to serve_asgi.
• A client with no Last-Event-ID starts from the oldest event kept, so a new browser receives up to the last hour first. One that is not an event's id gets 400.
• Each client sees every event of those types from its position. Filter on the client, or give each kind of client its own event type.