Python Patterns - Server-Sent Events

I have a love/hate relationship with live updates on web applications, partly because they’re usually a hassle to get right and partly due to their often needing major re-plumbing to work at all.

If you’re just tuning in, this is the second post in a somewhat erratic series that I’ve been meaning to write.

Most people reach for socket.io or WebSockets for various reasons, but the truth is that unless you’re trying to do something like collaborative editing or real-time multi-player gaming, they’re overkill and a royal pain to get working properly with both your target audience and your current web server and back-end stack. Furthermore, 95% of the time these things are used for _uni_directional communication.

Which is why I much prefer Server-Sent Events – it works fabulously for sending continuous updates to most browsers, and with the right polyfill it will fall back to polling even on extreme conditions like the server (or network) going away and returning.

Of course you still need to maintain a live connection for each client, etc., but at least it’s standard HTTP and you don’t have to fiddle with a separate server¹.

I’ve had a lot of fun playing around with live updating dashboards for kicks in Go and Java, but of late I’ve wanted a simple, straightforward Python-based solution I could integrate with on my standard project template (which, incidentally, I’ve been tweaking on Github).

And this week I’ve been toying with the notion of building a front-end to our room reservation system using Android devices, so I decided to jot down a few notes on how I managed to provide live updates to those.

First off, how do we cope with persistent connections? Well, that’s easy enough: I use uWSGI with gevent workers (or gunicorn on older setups), but getting a standalone Bottle server going for development is as easy as:

from bottle import run
from gevent import monkey; monkey.patch_all()

# ... your code here

run(server="gevent")

Now for the clever bits. First off, we need to pack replies in header: value format, with events separated by an extra newline:

def sse_pack(d):
    """Pack data in SSE format"""
    buffer = ''
    for k in ['retry','id','event','data']:
        if k in d.keys():
            buffer += '%s: %s\n' % (k, d[k])
    return buffer + '\n'

Next we need a way to keep sending events while the connection is open.

Fortunately, Bottle running under gevent will gleefully take a Python generator and keep piping whatever you yield into the open socket, so the rest is merely a matter of incrementing event IDs and keeping state consistent:

import json
from bottle import request, response, get, route

@get("/stream")
def stream_generator():
    # Keep event IDs consistent
    event_id = 0
    if 'Last-Event-Id' in request.headers:
        event_id = int(request.headers['Last-Event-Id']) + 1

    # Set up our message payload with a retry value in case of connection failure
    # (that's also the polling interval to be used as fallback by our polyfill)
    msg = {
        'retry': '2000'
    }

    # Provide an initial data dump to each new client
    response.headers['content-type'] = 'text/event-stream'
    response.headers['Access-Control-Allow-Origin'] = '*'
    msg.update({
         'event': 'init',
         'data' : json.dumps(get_current_shared_state()),
         'id'   : event_id
    })
    yield sse_pack(msg)

    # Now give them deltas as they arrive (say, from a message broker)
    event_id += 1    
    msg['event'] = 'delta'
    while True:
        # block until you get new data (from a queue, pub/sub, zmq, etc.)
        msg.update({
             'event': 'delta',
             'data' : json.dumps(queue.recv()),
             'id'   : event_id
        })
        yield sse_pack(msg)
        event_id += 1

I’ve done this kind of thing so far with Redis pubsub, MQTT brokers, 0MQ sockets, you name it – you might need to do a little more work if you’re not getting your client updates from a blocking mechanism, but this is the gist of things.

But what if you’re providing this stream to a statically-hosted single-page app?

(As I was, yesterday afternoon, live editing the HTML on MEO Cloud and gathering events from my laptop.)

Well, that’s easy enough: CORS-compliant browsers will issue an OPTIONS request to your server before actually requesting the event stream, so we can tell them it’s OK to have requests come in from pages hosted in other domains and specify which headers they are allowed to use:

@route("/stream", method="OPTIONS")
def options():
    response.headers.update({
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Methods': 'GET, OPTIONS',
        'Access-Control-Allow-Headers': 'X-REQUESTED-WITH, CACHE-CONTROL, LAST-EVENT-ID',
        'Content-Type': 'text/plain'
    })
    return ''

In case you’re curious, those are the headers the Android 4.2 stock WebView requires for EventSource to work in this scenario.

The JavaScript Side of Things

That’s a bit out of scope here, but here’s the gist of things:

function init(conf, app) {
    var source = new EventSource(conf.url)

    if (conf.debug) console.log("Binding event source");

    source.addEventListener('init', function(e) {
        app.trigger("model:init", e.data);
    }, false);

    source.addEventListener('delta', function(e) {
        app.trigger("view:update", e.data);
    }, false);

    source.addEventListener('error', function(e) {
        if (e.readyState == EventSource.CLOSED) {
            app.trigger("view:networkError");
        }
        else if( e.readyState == EventSource.OPEN) {
            app.trigger("feedback:connecting");
        }
    }, false);
}

As you can see, it’s mostly a matter of matching server-sent events to my single-page app triggers.

The above is using RiotJS, which I’m rather partial to these days because I can build a simple single-page app using it and Zepto with templating, a sane MVC approach and observables that fits completely under 50KB of code², so I’m sticking to it for simple, elegant stuff.

And that’s it for tonight. Next up, functional patterns or sane approaches to threading, depending on what I have on my plate.