run-dev: Use tornado for dev proxy server for HTTP requests.

- Use tornado as proxy server for development environment,
  replacing twisted (this doesn't support websockets).
- Upgrade tornado version to 4.4.1 (needs to be coupled to the
  above since neither change works without the other)

requirements/common.txt

@@ -21,6 +21,9 @@ SQLAlchemy==1.1.1
 # Needed for Tornado >3.2 compatibility
 backports-abc==0.4
 
+# Needed for Tornado 4 compatibility
+backports.ssl-match-hostname==3.5.0.1
+
 # Needed for S3 file uploads
 boto==2.42.0
 
@@ -127,7 +130,7 @@ sourcemap==0.1.8
 statsd==3.2.1
 
 # Tornado used for server->client push system
-tornado==3.2.2
+tornado==4.4.2
 
 # Needed for Python static typing
 typing==3.5.2.2

tools/run-dev.py

@@ -2,27 +2,23 @@
 from __future__ import print_function
 
 import optparse
-import pwd
-import subprocess
-import signal
-import traceback
-import sys
 import os
+import pwd
+import signal
+import subprocess
+import sys
+import time
+import traceback
 
-if False: from typing import Any
+from six.moves.urllib.parse import urlunparse
 
-from twisted.internet import reactor
-from twisted.web      import proxy, server, resource
+from tornado import httpclient
+from tornado import httputil
+from tornado import gen
+from tornado import web
+from tornado.ioloop import IOLoop
 
-# Monkey-patch twisted.web.http to avoid request.finish exceptions
-# https://trac.zulip.net/ticket/1728
-from twisted.web.http import Request
-orig_finish = Request.finish
-def patched_finish(self):
-    # type: (Any) -> None
-    if not self._disconnected:
-        orig_finish(self)
-Request.finish = patched_finish
+if False: from typing import Any, Callable, Generator, Optional
 
 if 'posix' in os.name and os.geteuid() == 0:
     raise RuntimeError("run-dev.py should not be run as root.")
@@ -53,7 +49,7 @@ parser.add_option('--no-clear-memcached',
                   action='store_false', dest='clear_memcached',
                   default=True, help='Do not clear memcached')
 
-(options, args) = parser.parse_args()
+(options, arguments) = parser.parse_args()
 
 if options.interface is None:
     user_id = os.getuid()
@@ -126,25 +122,157 @@ else:
 for cmd in cmds:
     subprocess.Popen(cmd)
 
-class Resource(resource.Resource):
-    def getChild(self, name, request):
-        # type: (bytes, server.Request) -> resource.Resource
 
-        # Assume an HTTP 1.1 request
-        proxy_host = request.requestHeaders.getRawHeaders('Host')
-        request.requestHeaders.setRawHeaders('X-Forwarded-Host', proxy_host)
-        if (request.uri.startswith(b'/json/events') or
-            request.uri.startswith(b'/api/v1/events') or
-            request.uri.startswith(b'/sockjs')):
-            return proxy.ReverseProxyResource('127.0.0.1', tornado_port, b'/' + name)
+def transform_url(protocol, path, query, target_port, target_host):
+    # type: (str, str, str, int, str) -> str
+    # generate url with target host
+    host = ":".join((target_host, str(target_port)))
+    newpath = urlunparse((protocol, host, path, '', query, ''))
+    return newpath
 
-        elif (request.uri.startswith(b'/webpack') or
-              request.uri.startswith(b'/socket.io')):
-            return proxy.ReverseProxyResource('127.0.0.1', webpack_port, b'/' + name)
 
-        return proxy.ReverseProxyResource('127.0.0.1', django_port, b'/'+name)
+@gen.engine
+def fetch_request(url, callback, **kwargs):
+    # type: (str, Any, **Any) -> Generator[Callable[..., Any], Any, None]
+    # use large timeouts to handle polling requests
+    req = httpclient.HTTPRequest(url, connect_timeout=240.0, request_timeout=240.0, **kwargs)
+    client = httpclient.AsyncHTTPClient()
+    # wait for response
+    response = yield gen.Task(client.fetch, req)
+    callback(response)
 
 
+class BaseHandler(web.RequestHandler):
+    # target server ip
+    target_host = '127.0.0.1'  # type: str
+    # target server port
+    target_port = None  # type: int
+
+    def _add_request_headers(self, exclude_lower_headers_list=None):
+        # type: (Optional[List[str]]) -> httputil.HTTPHeaders
+        exclude_lower_headers_list = exclude_lower_headers_list or []
+        headers = httputil.HTTPHeaders()
+        for header, v in self.request.headers.get_all():
+            if header.lower() not in exclude_lower_headers_list:
+                headers.add(header, v)
+        return headers
+
+    def get(self):
+        # type: () -> None
+        pass
+
+    def head(self):
+        # type: () -> None
+        pass
+
+    def post(self):
+        # type: () -> None
+        pass
+
+    def put(self):
+        # type: () -> None
+        pass
+
+    def patch(self):
+        # type: () -> None
+        pass
+
+    def options(self):
+        # type: () -> None
+        pass
+
+    def delete(self):
+        # type: () -> None
+        pass
+
+    def handle_response(self, response):
+        # type: (Any) -> None
+        if response.error and not isinstance(response.error, httpclient.HTTPError):
+            self.set_status(500)
+            self.write('Internal server error:\n' + str(response.error))
+        else:
+            self.set_status(response.code, response.reason)
+            self._headers = httputil.HTTPHeaders()  # clear tornado default header
+
+            for header, v in response.headers.get_all():
+                if header != 'Content-Length':
+                    # some header appear multiple times, eg 'Set-Cookie'
+                    self.add_header(header, v)
+            if response.body:
+                # rewrite Content-Length Header by the response
+                self.set_header('Content-Length', len(response.body))
+                self.write(response.body)
+        self.finish()
+
+    @web.asynchronous
+    def prepare(self):
+        # type: () -> None
+        url = transform_url(
+            self.request.protocol,
+            self.request.path,
+            self.request.query,
+            self.target_port,
+            self.target_host,
+        )
+        try:
+            fetch_request(
+                url=url,
+                callback=self.handle_response,
+                method=self.request.method,
+                headers=self._add_request_headers(["upgrade-insecure-requests"]),
+                follow_redirects=False,
+                body=getattr(self.request, 'body'),
+                allow_nonstandard_methods=True
+            )
+        except httpclient.HTTPError as e:
+            if hasattr(e, 'response') and e.response:
+                self.handle_response(e.response)
+            else:
+                self.set_status(500)
+                self.write('Internal server error:\n' + str(e))
+                self.finish()
+
+
+class WebPackHandler(BaseHandler):
+    target_port = webpack_port
+
+
+class DjangoHandler(BaseHandler):
+    target_port = django_port
+
+
+class TornadoHandler(BaseHandler):
+    target_port = tornado_port
+
+
+class Application(web.Application):
+    def __init__(self):
+        # type: () -> None
+        handlers = [
+            (r"/sockjs/.*/websocket$", TornadoHandler),
+            (r"/json/events.*", TornadoHandler),
+            (r"/api/v1/events.*", TornadoHandler),
+            (r"/webpack.*", WebPackHandler),
+            (r"/sockjs.*", TornadoHandler),
+            (r"/socket.io.*", WebPackHandler),
+            (r"/.*", DjangoHandler)
+        ]
+        super(Application, self).__init__(handlers)
+
+
+def on_shutdown():
+    # type: () -> None
+    IOLoop.instance().stop()
+
+
+def shutdown_handler(*args, **kwargs):
+    # type: (*Any, **Any) -> None
+    io_loop = IOLoop.instance()
+    if io_loop._callbacks:
+        io_loop.add_timeout(time.time() + 1, shutdown_handler)
+    else:
+        io_loop.stop()
+
 # log which services/ports will be started
 print("Starting Zulip services on ports: web proxy: {},".format(proxy_port),
       "Django: {}, Tornado: {}".format(django_port, tornado_port), end='')
@@ -153,11 +281,17 @@ class Resource(resource.Resource):
 else:
     print(", webpack: {}".format(webpack_port))
 
-    print(WARNING + "Note: only port {} is exposed to the host in a Vagrant environment.".format(proxy_port) + ENDC)
+print("".join((WARNING,
+               "Note: only port {} is exposed to the host in a Vagrant environment.".format(
+                   proxy_port), ENDC)))
 
 try:
-    reactor.listenTCP(proxy_port, server.Site(Resource()), interface=options.interface)
-    reactor.run()
+    app = Application()
+    app.listen(proxy_port)
+    ioloop = IOLoop.instance()
+    for s in (signal.SIGINT, signal.SIGTERM):
+        signal.signal(s, shutdown_handler)
+    ioloop.start()
 except:
     # Print the traceback before we get SIGTERM and die.
     traceback.print_exc()

version.py

@@ -1,2 +1,2 @@
 ZULIP_VERSION = "1.4.1+git"
-PROVISION_VERSION = '1.00'
+PROVISION_VERSION = '2.0.0'