Skip to content

Instantly share code, notes, and snippets.

Last active February 8, 2024 02:17
Show Gist options
  • Save mivade/d474e0540036d873047f to your computer and use it in GitHub Desktop.
Save mivade/d474e0540036d873047f to your computer and use it in GitHub Desktop.
Tornado server-sent events
"""Demonstration of server-sent events with Tornado. To see the
stream, you can either point your browser to ``http://localhost:8080``
or use ``curl`` like so::
$ curl http://localhost:8080/events
import signal
from tornado import web, gen
from tornado.options import options
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop, PeriodicCallback
from tornado.iostream import StreamClosedError
html = """
<div id="messages"></div>
<script type="text/javascript">
var source = new EventSource('/events');
source.onmessage = function(message) {
var div = document.getElementById("messages");
div.innerHTML = + "<br>" + div.innerHTML;
def fibonacci():
a, b = 0, 1
while True:
yield a
a, b = b, a + b
class DataSource(object):
"""Generic object for producing data to feed to clients."""
def __init__(self, initial_data=None):
self._data = initial_data
def data(self):
return self._data
def data(self, new_data):
self._data = new_data
class EventSource(web.RequestHandler):
"""Basic handler for server-sent events."""
def initialize(self, source):
"""The ``source`` parameter is a string that is updated with
new data. The :class:`EventSouce` instance will continuously
check if it is updated and publish to clients when it is.
self.source = source
self._last = None
self.set_header('content-type', 'text/event-stream')
self.set_header('cache-control', 'no-cache')
def publish(self, data):
"""Pushes data to a listener."""
self.write('data: {}\n\n'.format(data))
yield self.flush()
except StreamClosedError:
def get(self):
while True:
if != self._last:
yield self.publish(
self._last =
yield gen.sleep(0.005)
class MainHandler(web.RequestHandler):
def get(self):
if __name__ == "__main__":
generator = fibonacci()
publisher = DataSource(next(generator))
def get_next(): = next(generator)
checker = PeriodicCallback(lambda: get_next(), 1000.)
app = web.Application(
(r'/', MainHandler),
(r'/events', EventSource, dict(source=publisher))
server = HTTPServer(app)
signal.signal(signal.SIGINT, lambda x, y: IOLoop.instance().stop())
Copy link

BTW why you were using , "\n" in
self.write('data:` {}\n\n'.format(data))


Copy link

mivade commented Apr 10, 2020

The \n\n is how messages are delimited with SSE.

Copy link

Prabhusarathy18 commented Apr 12, 2020

The \n\n is how messages are delimited with SSE.

But what if i need to print xx in one line and yy in another line.

Whenever i send strings with xx\nyy it neglects those \n

Could u help me out?

Copy link

ChouJoe commented Jun 23, 2020

hello, since EventSource def get() function include While True, i am confused with how to close the SSE connect?

Copy link

mivade commented Jun 23, 2020

This demo was assuming there's always going to be data streaming. If you have a condition where you're done sending, you can just return from the function (or raise web.Finish).

Copy link

gruentee commented Oct 7, 2020

Thanks for this useful demo!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment