WIP: Streaming

sanic/app.py

@@ -22,7 +22,7 @@
 from sanic.exceptions import SanicException, ServerError, URLBuildError
 from sanic.handlers import ErrorHandler
 from sanic.log import LOGGING_CONFIG_DEFAULTS, error_logger, logger
-from sanic.response import HTTPResponse, StreamingHTTPResponse
+from sanic.response import BaseHTTPResponse, HTTPResponse
 from sanic.router import Router
 from sanic.server import (
     AsyncioServer,
@@ -82,7 +82,6 @@ def __init__(
         self.strict_slashes = strict_slashes
         self.listeners = defaultdict(list)
         self.is_running = False
-        self.is_request_stream = False
         self.websocket_enabled = False
         self.websocket_tasks = set()
         self.named_request_middleware = {}
@@ -187,9 +186,6 @@ def route(
         if not uri.startswith("/"):
             uri = "/" + uri
 
-        if stream:
-            self.is_request_stream = True
-
         if strict_slashes is None:
             strict_slashes = self.strict_slashes
 
@@ -942,7 +938,41 @@ def converted_response_type(self, response):
         """
         pass
 
-    async def handle_request(self, request, write_callback, stream_callback):
+    async def handle_exception(self, request, exception):
+        try:
+            response = self.error_handler.response(request, exception)
+            if isawaitable(response):
+                response = await response
+        except Exception as e:
+            if isinstance(e, SanicException):
+                response = self.error_handler.default(request, e)
+            elif self.debug:
+                response = HTTPResponse(
+                    f"Error while handling error: {e}\nStack: {format_exc()}",
+                    status=500,
+                )
+            else:
+                response = HTTPResponse(
+                    "An error occurred while handling an error", status=500
+                )
+        if response is not None:
+            try:
+                response = await request.respond(response)
+            except BaseException:
+                # Skip response middleware
+                request.stream.respond(response)
+                await response.send(end_stream=True)
+                raise
+        else:
+            response = request.stream.response
+        if isinstance(response, BaseHTTPResponse):
+            await response.send(end_stream=True)
+        else:
+            raise ServerError(
+                f"Invalid response type {response!r} (need HTTPResponse)"
+            )
+
+    async def handle_request(self, request):
         """Take a request from the HTTP Server and return a response object
         to be sent back The HTTP Server only expects a response object, so
         exception handling must be done here
@@ -958,11 +988,19 @@ async def handle_request(self, request, write_callback, stream_callback):
         # Define `response` var here to remove warnings about
         # allocation before assignment below.
         response = None
-        cancelled = False
         name = None
         try:
             # Fetch handler from router
             handler, args, kwargs, uri, name = self.router.get(request)
+            request.name = name
+
+            if request.stream.request_body:
+                if self.router.is_stream_handler(request):
+                    # Streaming handler: lift the size limit
+                    request.stream.request_max_size = float("inf")
+                else:
+                    # Non-streaming handler: preload body
+                    await request.receive_body()
 
             # -------------------------------------------- #
             # Request Middleware
@@ -999,73 +1037,25 @@ async def handle_request(self, request, write_callback, stream_callback):
                 response = handler(request, *args, **kwargs)
                 if isawaitable(response):
                     response = await response
+            if response:
+                response = await request.respond(response)
+            else:
+                response = request.stream.response
+            # Make sure that response is finished / run StreamingHTTP callback
+            if isinstance(response, BaseHTTPResponse):
+                await response.send(end_stream=True)
+            else:
+                raise ServerError(
+                    f"Invalid response type {response!r} (need HTTPResponse)"
+                )
+
         except CancelledError:
-            # If response handler times out, the server handles the error
-            # and cancels the handle_request job.
-            # In this case, the transport is already closed and we cannot
-            # issue a response.
-            response = None
-            cancelled = True
+            raise
         except Exception as e:
             # -------------------------------------------- #
             # Response Generation Failed
             # -------------------------------------------- #
-
-            try:
-                response = self.error_handler.response(request, e)
-                if isawaitable(response):
-                    response = await response
-            except Exception as e:
-                if isinstance(e, SanicException):
-                    response = self.error_handler.default(
-                        request=request, exception=e
-                    )
-                elif self.debug:
-                    response = HTTPResponse(
-                        "Error while handling error: {}\nStack: {}".format(
-                            e, format_exc()
-                        ),
-                        status=500,
-                    )
-                else:
-                    response = HTTPResponse(
-                        "An error occurred while handling an error", status=500
-                    )
-        finally:
-            # -------------------------------------------- #
-            # Response Middleware
-            # -------------------------------------------- #
-            # Don't run response middleware if response is None
-            if response is not None:
-                try:
-                    response = await self._run_response_middleware(
-                        request, response, request_name=name
-                    )
-                except CancelledError:
-                    # Response middleware can timeout too, as above.
-                    response = None
-                    cancelled = True
-                except BaseException:
-                    error_logger.exception(
-                        "Exception occurred in one of response "
-                        "middleware handlers"
-                    )
-            if cancelled:
-                raise CancelledError()
-
-        # pass the response to the correct callback
-        if write_callback is None or isinstance(
-            response, StreamingHTTPResponse
-        ):
-            if stream_callback:
-                await stream_callback(response)
-            else:
-                # Should only end here IF it is an ASGI websocket.
-                # TODO:
-                # - Add exception handling
-                pass
-        else:
-            write_callback(response)
+            await self.handle_exception(request, e)
 
     # -------------------------------------------------------------------- #
     # Testing
@@ -1346,6 +1336,8 @@ async def _run_response_middleware(
                     _response = await _response
                 if _response:
                     response = _response
+                    if isinstance(response, BaseHTTPResponse):
+                        response = request.stream.respond(response)
                     break
         return response
 
@@ -1395,7 +1387,7 @@ def _helper(
         server_settings = {
             "protocol": protocol,
             "request_class": self.request_class,
-            "is_request_stream": self.is_request_stream,
+            "is_request_stream": True,
             "router": self.router,
             "host": host,
             "port": port,