Server-sent events with Python,Twisted and Flask: is this a correct approach for sleeping?

StackOverflow https://stackoverflow.com/questions/20380066

Question

I started looking at server-sent events and got interested in trying them out with my preferred tools, Python, Flask and Twisted. I'm asking if sleeping the way i'm doing it is fine, compared to the gevent's greenlet.sleep way of doing, this is my very simple code taken and "ported" to Twisted (from gevent):

#!/usr/bin/env python

import random
from twisted.web.server import Site
from twisted.web.wsgi import WSGIResource
from twisted.internet import reactor
import time

from flask import Flask, request, Response
app = Flask(__name__)

def event_stream():
    count = 0
    while True:
        count += 1
        yield 'data: %c (%d)\n\n' % (random.choice('abcde'), count)
        time.sleep(1)


@app.route('/my_event_source')
def sse_request():
    return Response(
            event_stream(),
            mimetype='text/event-stream')


@app.route('/')
def page():
    return '''
<!DOCTYPE html>
<html>
    <head>
        <script type="text/javascript" src="//code.jquery.com/jquery-1.8.0.min.js"></script>
        <script type="text/javascript">
            $(document).ready(
                    function() {
                        sse = new EventSource('/my_event_source');
                        sse.onmessage = function(message) {
                            console.log('A message has arrived!');
                            $('#output').append('<li>'+message.data+'</li>');
                        }

                    })
        </script>
    </head>
    <body>
        <h2>Demo</h2>
        <ul id="output"></ul>
    </body>
</html>
'''


if __name__ == '__main__':
    resource = WSGIResource(reactor, reactor.getThreadPool(), app)
    site = Site(resource)
    reactor.listenTCP(8001, site)
    reactor.run()

Although time.sleep is a blocking function, that will not block the Twisted reactor, and this should be proven by the fact that multiple different browsers can access the page and receive the event properly: using different browsers is needed in case, as Chromium is doing, multiple different requests with the same URI will get queued and since this is a streaming response, that browser queue will be busy until the socket or the request are closed. What are your toughts? Any better way? There is not much sample code about this with Twisted and Flask around..

Was it helpful?

Solution

Your example uses twisted only as a wsgi container. As well as any other thread-based wsgi container it allows you to use time.sleep(1).

It is the case where allowing twisted to handle /my_event_source directly might be beneficial. Here's an example from Using server sent events implemented in Python using twisted:

def cycle(echo):
    # Every second, sent a "ping" event.
    timestr = datetime.utcnow().isoformat()+"Z"
    echo("event: ping\n")
    echo('data: ' +  json.dumps(dict(time=timestr)))
    echo("\n\n")

    # Send a simple message at random intervals.
    if random.random() < 0.1:
        echo("data: This is a message at time {}\n\n".format(timestr))

class SSEResource(resource.Resource):
    def render_GET(self, request):
        request.setHeader("Content-Type", "text/event-stream")
        lc = task.LoopingCall(cycle, request.write)
        lc.start(1) # repeat every second
        request.notifyFinish().addBoth(lambda _: lc.stop())
        return server.NOT_DONE_YET

where the client static/index.html is from the same source:

<!doctype html>
<title>Using server-sent events</title>
<ol id="eventlist">nothing sent yet.</ol>
<script>
if (!!window.EventSource) {
  var eventList = document.getElementById("eventlist");
  var source = new EventSource('/my_event_source');
  source.onmessage = function(e) {
    var newElement = document.createElement("li");

    newElement.innerHTML = "message: " + e.data;
    eventList.appendChild(newElement);
  }
  source.addEventListener("ping", function(e) {
    var newElement = document.createElement("li");

    var obj = JSON.parse(e.data);
    newElement.innerHTML = "ping at " + obj.time;
    eventList.appendChild(newElement);
  }, false);
  source.onerror = function(e) {
    alert("EventSource failed.");
    source.close();
  };
}
</script>

You could combine it with your wsgi application:

app = Flask(__name__)
@app.route('/')
def index():
    return redirect(url_for('static', filename='index.html'))

if __name__ == "__main__":
    root = resource.Resource()
    root.putChild('', wsgi.WSGIResource(reactor, reactor.getThreadPool(), app))
    root.putChild('static', static.File("./static"))
    root.putChild('my_event_source', SSEResource())

    reactor.listenTCP(8001, server.Site(root))
    reactor.run()

WSGIResource expects to handle all urls so the routing code needs to be rewritten to support multiple flask handlers.

Licensed under: CC-BY-SA with attribution
Not affiliated with StackOverflow
scroll top