diff --git a/sanic/app.py b/sanic/app.py index 2efe32c4..998fc3cb 100644 --- a/sanic/app.py +++ b/sanic/app.py @@ -13,7 +13,7 @@ from sanic.constants import HTTP_METHODS from sanic.exceptions import ServerError, URLBuildError, SanicException from sanic.handlers import ErrorHandler from sanic.log import log -from sanic.response import HTTPResponse +from sanic.response import HTTPResponse, StreamingHTTPResponse from sanic.router import Router from sanic.server import serve, serve_multiple, HttpProtocol from sanic.static import register as static_register @@ -342,14 +342,17 @@ class Sanic: def converted_response_type(self, response): pass - async def handle_request(self, request, response_callback): + async def handle_request(self, request, write_callback, stream_callback): """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 :param request: HTTP Request object - :param response_callback: Response function to be called with the - response as the only argument + :param write_callback: Synchronous response function to be + called with the response as the only argument + :param stream_callback: Coroutine that handles streaming a + StreamingHTTPResponse if produced by the handler. + :return: Nothing """ try: @@ -416,7 +419,11 @@ class Sanic: response = HTTPResponse( "An error occurred while handling an error") - await response_callback(response) + # pass the response to the correct callback + if isinstance(response, StreamingHTTPResponse): + await stream_callback(response) + else: + write_callback(response) # -------------------------------------------------------------------- # # Testing diff --git a/sanic/server.py b/sanic/server.py index e1f68935..5ae9e937 100644 --- a/sanic/server.py +++ b/sanic/server.py @@ -162,27 +162,61 @@ class HttpProtocol(asyncio.Protocol): self.request.body = b''.join(self.request.body) self._request_handler_task = self.loop.create_task( - self.request_handler(self.request, self.write_response)) + self.request_handler( + self.request, + self.write_response, + self.stream_response)) # -------------------------------------------- # # Responding # -------------------------------------------- # - async def write_response(self, response): + def write_response(self, response): + """ + Writes response content synchronously to the transport. + """ try: keep_alive = ( self.parser.should_keep_alive() and not self.signal.stopped) - if isinstance(response, StreamingHTTPResponse): - # streaming responses should have direct write access to the - # transport - response.transport = self.transport - await response.stream( - self.request.version, keep_alive, self.request_timeout) + self.transport.write( + response.output( + self.request.version, keep_alive, + self.request_timeout)) + except AttributeError: + log.error( + ('Invalid response object for url {}, ' + 'Expected Type: HTTPResponse, Actual Type: {}').format( + self.url, type(response))) + self.write_error(ServerError('Invalid response type')) + except RuntimeError: + log.error( + 'Connection lost before response written @ {}'.format( + self.request.ip)) + except Exception as e: + self.bail_out( + "Writing response failed, connection closed {}".format( + repr(e))) + finally: + if not keep_alive: + self.transport.close() else: - self.transport.write( - response.output( - self.request.version, keep_alive, - self.request_timeout)) + self._last_request_time = current_time + self.cleanup() + + async def stream_response(self, response): + """ + Streams a response to the client asynchronously. Attaches + the transport to the response so the response consumer can + write to the response as needed. + """ + + try: + keep_alive = ( + self.parser.should_keep_alive() and not self.signal.stopped) + + response.transport = self.transport + await response.stream( + self.request.version, keep_alive, self.request_timeout) except AttributeError: log.error( ('Invalid response object for url {}, '