refactor, expose app object to **server

tests/http_server.py

@@ -68,6 +68,7 @@ async def on_connect(**server):
 @app.on_close
 async def on_close(**server):
     assert server['context'].foo == 'bar'
+    server['globals'].options['max_queue_size'] = 128
 
     return True
 

tremolo/asgi_server.py

@@ -23,12 +23,12 @@
 class ASGIServer(HTTPProtocol):
     __slots__ = ('response', '_scope', '_read', '_websocket', '_waiter')
 
-    def __init__(self, context, **kwargs):
-        super().__init__(context, **kwargs)
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
 
         self._scope = {
             'asgi': {'version': '3.0', 'spec_version': '2.3'},
-            'root_path': self.options['_root_path'],
+            'root_path': self.options['root_path'],
             'server': self.globals.info['server']
         }
         self.response = None  # set in headers_received
@@ -77,7 +77,7 @@ async def headers_received(self, response):
             self._handle_http()
 
         try:
-            await self.options['_app'](self._scope, self.receive, self.send)
+            await self.options['app'](self._scope, self.receive, self.send)
 
             if not self._waiter.done():
                 self.logger.info('handler exited early (no close?)')

tremolo/http_server.py

@@ -9,17 +9,18 @@
 class HTTPServer(HTTPProtocol):
     __slots__ = ('_routes', '_middlewares', '_server')
 
-    def __init__(self, context, _routes=None, _middlewares=None, **kwargs):
-        super().__init__(context, **kwargs)
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
 
-        self._routes = _routes
-        self._middlewares = _middlewares
+        self._routes = self.app.routes
+        self._middlewares = self.app.middlewares
         self._server = {
             'response': None,  # set in headers_received
-            'globals': context,
+            'globals': self.globals,
             'context': self.context,
-            'loop': kwargs['loop'],
-            'logger': kwargs['logger'],
+            'app': self.app,
+            'loop': self.loop,
+            'logger': self.logger,
             'lock': kwargs['lock']
         }
 
@@ -48,14 +49,14 @@ def connection_made(self, transport):
         super().connection_made(transport)
 
         if self._middlewares['connect']:
-            self.context.ON_CONNECT = self.create_background_task(
+            self.context.ON_CONNECT = self.app.create_task(
                 self._connection_made()
             )
             self.add_close_callback(self.context.ON_CONNECT.cancel)
 
     def connection_lost(self, exc):
         if self._middlewares['close']:
-            task = self.create_background_task(self._connection_lost(exc))
+            task = self.app.create_task(self._connection_lost(exc))
             self.loop.call_at(
                 self.loop.time() + self.options['app_close_timeout'],
                 task.cancel

tremolo/lib/http_protocol.py

@@ -20,6 +20,7 @@ class HTTPProtocol(asyncio.Protocol):
     __slots__ = ('globals',
                  'context',
                  'options',
+                 'app',
                  'loop',
                  'logger',
                  'fileno',
@@ -30,12 +31,13 @@ class HTTPProtocol(asyncio.Protocol):
                  '_header_buf',
                  '_waiters')
 
-    def __init__(self, context, loop=None, logger=None, **kwargs):
-        self.globals = context  # a worker-level context
+    def __init__(self, app, **kwargs):
+        self.globals = app.context  # a worker-level context
         self.context = ConnectionContext()
-        self.options = kwargs
-        self.loop = loop
-        self.logger = logger
+        self.options = kwargs['options']
+        self.app = app
+        self.loop = app.loop
+        self.logger = app.logger
         self.fileno = -1
         self.queue = None
         self.handlers = set()
@@ -56,14 +58,6 @@ def tasks(self):
     def add_close_callback(self, callback):
         self.tasks.add(callback)
 
-    def create_background_task(self, coro):
-        task = self.loop.create_task(coro)
-
-        self.globals.tasks.add(task)
-        task.add_done_callback(self.globals.tasks.discard)
-
-        return task
-
     def create_task(self, coro):
         task = self.loop.create_task(coro)
 
@@ -92,10 +86,8 @@ def connection_made(self, transport):
 
         self._waiters['request'] = self.loop.create_future()
 
-        self.add_close_callback(
-            self.create_background_task(self._send_data()).cancel
-        )
-        self.add_close_callback(self.create_background_task(
+        self.add_close_callback(self.app.create_task(self._send_data()).cancel)
+        self.add_close_callback(self.app.create_task(
             self.set_timeout(self._waiters['request'],
                              timeout=self.options['request_timeout'],
                              timeout_cb=self.request_timeout)
@@ -334,9 +326,8 @@ def data_received(self, data):
                                     excludes=[b'proxy'])
 
                 if header.is_request:
-                    task = self.create_background_task(
-                        self._handle_request(header)
-                    )
+                    task = self.app.create_task(self._handle_request(header))
+
                     self.handlers.add(task)
                     task.add_done_callback(self.handlers.discard)
                 else:
@@ -457,7 +448,7 @@ async def _handle_keepalive(self):
 
             self._waiters['request'] = self.loop.create_future()
 
-            self.add_close_callback(self.create_background_task(
+            self.add_close_callback(self.app.create_task(
                 self.set_timeout(self._waiters['request'],
                                  timeout=self.options['keepalive_timeout'],
                                  timeout_cb=self.keepalive_timeout)