Tornado 5 support

README.rst

@@ -11,7 +11,7 @@ Features
 
 - Fully compatible with the Javascript `engine.io-client`_ library, versions 1.5.0 and up.
 - Compatible with Python 2.7 and Python 3.3+.
-- Supports large number of clients even on modest hardware when used with an asynchronous server based on `asyncio`_(`sanic`_ or `aiohttp`_), `eventlet`_ or `gevent`_. For development and testing, any WSGI compliant multi-threaded server can be used.
+- Supports large number of clients even on modest hardware when used with an asynchronous server based on `asyncio`_(`sanic`_, `aiohttp`_ or `tornado`_), `eventlet`_ or `gevent`_. For development and testing, any WSGI compliant multi-threaded server can be used.
 - Includes a WSGI middleware that integrates Engine.IO traffic with standard WSGI applications.
 - Uses an event-based architecture implemented with decorators that hides the details of the protocol.
 - Implements HTTP long-polling and WebSocket transports.
@@ -111,6 +111,7 @@ Resources
 .. _asyncio: https://docs.python.org/3/library/asyncio.html
 .. _sanic: http://sanic.readthedocs.io/
 .. _aiohttp: http://aiohttp.readthedocs.io/
+.. _tornado: http://www.tornadoweb.org/
 .. _eventlet: http://eventlet.net/
 .. _gevent: http://gevent.org/
 .. _aiohttp: http://aiohttp.readthedocs.io/

engineio/__init__.py

@@ -4,11 +4,12 @@
 from .server import Server
 if sys.version_info >= (3, 5):  # pragma: no cover
     from .asyncio_server import AsyncServer
+    from .async_tornado import get_tornado_handler
 else:  # pragma: no cover
     AsyncServer = None
 
 __version__ = '2.1.1'
 
 __all__ = ['__version__', 'Middleware', 'Server']
 if AsyncServer is not None:  # pragma: no cover
-    __all__.append('AsyncServer')
+    __all__ += ['AsyncServer', 'get_tornado_handler']

engineio/async_aiohttp.py

@@ -72,7 +72,7 @@ def translate_request(request):
     return environ
 
 
-def make_response(status, headers, payload):
+def make_response(status, headers, payload, environ):
     """This function generates an appropriate response object for this async
     mode.
     """

engineio/async_sanic.py

@@ -89,7 +89,7 @@ async def read(self, length=None):
     return environ
 
 
-def make_response(status, headers, payload):
+def make_response(status, headers, payload, environ):
     """This function generates an appropriate response object for this async
     mode.
     """

engineio/async_tornado.py

@@ -0,0 +1,154 @@
+import asyncio
+import sys
+from urllib.parse import urlsplit
+
+try:
+    import tornado.web
+    import tornado.websocket
+except ImportError:
+    pass
+import six
+
+
+def get_tornado_handler(engineio_server):
+    class Handler(tornado.websocket.WebSocketHandler):
+        def __init__(self, *args, **kwargs):
+            super().__init__(*args, **kwargs)
+            self.receive_queue = asyncio.Queue()
+
+        async def get(self):
+            if self.request.headers.get('Upgrade', '').lower() == 'websocket':
+                super().get()
+            await engineio_server.handle_request(self)
+
+        async def post(self):
+            await engineio_server.handle_request(self)
+
+        async def options(self):
+            await engineio_server.handle_request(self)
+
+        async def on_message(self, message):
+            await self.receive_queue.put(message)
+
+        async def get_next_message(self):
+            return await self.receive_queue.get()
+
+        def on_close(self):
+            self.receive_queue.put_nowait(None)
+
+    return Handler
+
+
+def translate_request(handler):
+    """This function takes the arguments passed to the request handler and
+    uses them to generate a WSGI compatible environ dictionary.
+    """
+    class AwaitablePayload(object):
+        def __init__(self, payload):
+            self.payload = payload or b''
+
+        async def read(self, length=None):
+            if length is None:
+                r = self.payload
+                self.payload = b''
+            else:
+                r = self.payload[:length]
+                self.payload = self.payload[length:]
+            return r
+
+    payload = handler.request.body
+
+    uri_parts = urlsplit(handler.request.path)
+    environ = {
+        'wsgi.input': AwaitablePayload(payload),
+        'wsgi.errors': sys.stderr,
+        'wsgi.version': (1, 0),
+        'wsgi.async': True,
+        'wsgi.multithread': False,
+        'wsgi.multiprocess': False,
+        'wsgi.run_once': False,
+        'SERVER_SOFTWARE': 'aiohttp',
+        'REQUEST_METHOD': handler.request.method,
+        'QUERY_STRING': handler.request.query or '',
+        'RAW_URI': handler.request.path,
+        'SERVER_PROTOCOL': 'HTTP/%s' % handler.request.version,
+        'REMOTE_ADDR': '127.0.0.1',
+        'REMOTE_PORT': '0',
+        'SERVER_NAME': 'aiohttp',
+        'SERVER_PORT': '0',
+        'tornado.handler': handler
+    }
+
+    for hdr_name, hdr_value in handler.request.headers.items():
+        hdr_name = hdr_name.upper()
+        if hdr_name == 'CONTENT-TYPE':
+            environ['CONTENT_TYPE'] = hdr_value
+            continue
+        elif hdr_name == 'CONTENT-LENGTH':
+            environ['CONTENT_LENGTH'] = hdr_value
+            continue
+
+        key = 'HTTP_%s' % hdr_name.replace('-', '_')
+        if key in environ:
+            hdr_value = '%s,%s' % (environ[key], hdr_value)
+
+        environ[key] = hdr_value
+
+    environ['wsgi.url_scheme'] = environ.get('HTTP_X_FORWARDED_PROTO', 'http')
+
+    path_info = uri_parts.path
+
+    environ['PATH_INFO'] = path_info
+    environ['SCRIPT_NAME'] = ''
+
+    return environ
+
+
+def make_response(status, headers, payload, environ):
+    """This function generates an appropriate response object for this async
+    mode.
+    """
+    tornado_handler = environ['tornado.handler']
+    tornado_handler.set_status(int(status.split()[0]))
+    for header, value in headers:
+        tornado_handler.set_header(header, value)
+    tornado_handler.write(payload)
+    tornado_handler.finish()
+
+
+class WebSocket(object):  # pragma: no cover
+    """
+    This wrapper class provides a tornado WebSocket interface that is
+    somewhat compatible with eventlet's implementation.
+    """
+    def __init__(self, handler):
+        self.handler = handler
+        self.tornado_handler = None
+
+    async def __call__(self, environ):
+        self.tornado_handler = environ['tornado.handler']
+        self.environ = environ
+        await self.handler(self)
+
+    async def close(self):
+        self.tornado_handler.close()
+
+    async def send(self, message):
+        self.tornado_handler.write_message(
+            message, binary=isinstance(message, bytes))
+
+    async def wait(self):
+        msg = await self.tornado_handler.get_next_message()
+        if not isinstance(msg, six.binary_type) and \
+                not isinstance(msg, six.text_type):
+            raise IOError()
+        return msg
+
+
+_async = {
+    'asyncio': True,
+    'translate_request': translate_request,
+    'make_response': make_response,
+    'websocket': sys.modules[__name__],
+    'websocket_class': 'WebSocket'
+}

engineio/asyncio_server.py

@@ -18,9 +18,10 @@ class AsyncServer(server.Server):
 
     :param async_mode: The asynchronous model to use. See the Deployment
                        section in the documentation for a description of the
-                       available options. Valid async modes are "aiohttp". If
-                       this argument is not given, an async mode is chosen
-                       based on the installed packages.
+                       available options. Valid async modes are "aiohttp",
+                       "sanic" and "tornado". If this argument is not given,
+                       an async mode is chosen based on the installed
+                       packages.
     :param ping_timeout: The time in seconds that the client waits for the
                          server to respond before disconnecting.
     :param ping_interval: The interval in seconds at which the client pings
@@ -55,7 +56,7 @@ def is_asyncio_based(self):
         return True
 
     def async_modes(self):
-        return ['aiohttp', 'sanic']
+        return ['aiohttp', 'sanic', 'tornado']
 
     def attach(self, app, engineio_path='engine.io'):
         """Attach the Engine.IO server to an application."""
@@ -192,7 +193,8 @@ async def handle_request(self, *args, **kwargs):
         cors_headers = self._cors_headers(environ)
         return self._async['make_response'](r['status'],
                                             r['headers'] + cors_headers,
-                                            r['response'])
+                                            r['response'],
+                                            environ)
 
     def start_background_task(self, target, *args, **kwargs):
         """Start a background task using the appropriate async model.

examples/README.rst

@@ -18,3 +18,9 @@ sanic
 -----
 
 Examples that are compatible with the sanic framework for asyncio.
+
+
+tornado
+-------
+
+Examples that are compatible with the Tornado framework.

examples/tornado/README.rst

@@ -0,0 +1,39 @@
+Engine.IO Examples
+==================
+
+This directory contains example Engine.IO applications that are compatible
+with the Tornado framework. These applications require Tornado 5 and Python
+3.5 or later.
+
+simple.py
+---------
+
+A basic application in which the client sends messages to the server and the
+server responds.
+
+latency.py
+----------
+
+A port of the latency application included in the official Engine.IO
+Javascript server. In this application the client sends *ping* messages to
+the server, which are responded by the server with a *pong*. The client
+measures the time it takes for each of these exchanges and plots these in real
+time to the page.
+
+This is an ideal application to measure the performance of the different
+asynchronous modes supported by the Engine.IO server.
+
+Running the Examples
+--------------------
+
+To run these examples, create a virtual environment, install the requirements
+and then run::
+
+    $ python simple.py
+
+or::
+
+    $ python latency.py
+
+You can then access the application from your web browser at
+``http://localhost:8888``.

examples/tornado/latency.py

@@ -0,0 +1,41 @@
+import os
+
+import tornado.ioloop
+from tornado.options import define, options, parse_command_line
+import tornado.web
+
+import engineio
+
+define("port", default=8888, help="run on the given port", type=int)
+define("debug", default=False, help="run in debug mode")
+
+eio = engineio.AsyncServer(async_mode='tornado')
+
+
+class MainHandler(tornado.web.RequestHandler):
+    def get(self):
+        self.render("latency.html")
+
+
+@eio.on('message')
+async def message(sid, data):
+    await eio.send(sid, 'pong', binary=False)
+
+
+def main():
+    parse_command_line()
+    app = tornado.web.Application(
+        [
+            (r"/", MainHandler),
+            (r"/engine.io/", engineio.get_tornado_handler(eio)),
+        ],
+        template_path=os.path.join(os.path.dirname(__file__), "templates"),
+        static_path=os.path.join(os.path.dirname(__file__), "static"),
+        debug=options.debug,
+    )
+    app.listen(options.port)
+    tornado.ioloop.IOLoop.current().start()
+
+
+if __name__ == "__main__":
+    main()

examples/tornado/requirements.txt

@@ -0,0 +1,3 @@
+tornado==5.0.2
+python-engineio
+six==1.10.0

examples/tornado/simple.py

@@ -0,0 +1,53 @@
+import os
+
+import tornado.ioloop
+from tornado.options import define, options, parse_command_line
+import tornado.web
+
+import engineio
+from engineio.async_tornado import get_engineio_handler
+
+define("port", default=8888, help="run on the given port", type=int)
+define("debug", default=False, help="run in debug mode")
+
+eio = engineio.AsyncServer(async_mode='tornado')
+
+
+class MainHandler(tornado.web.RequestHandler):
+    def get(self):
+        self.render("simple.html")
+
+
+@eio.on('connect')
+def connect(sid, environ):
+    print("connect ", sid)
+
+
+@eio.on('message')
+async def message(sid, data):
+    print('message from', sid, data)
+    await eio.send(sid, 'Thank you for your message!', binary=False)
+
+
+@eio.on('disconnect')
+def disconnect(sid):
+    print('disconnect ', sid)
+
+
+def main():
+    parse_command_line()
+    app = tornado.web.Application(
+        [
+            (r"/", MainHandler),
+            (r"/engine.io/", engineio.get_tornado_handler(eio)),
+        ],
+        template_path=os.path.join(os.path.dirname(__file__), "templates"),
+        static_path=os.path.join(os.path.dirname(__file__), "static"),
+        debug=options.debug,
+    )
+    app.listen(options.port)
+    tornado.ioloop.IOLoop.current().start()
+
+
+if __name__ == "__main__":
+    main()