prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Memory usage streaming large responses
  We've been running into memory issues when providing very large async generators to a streaming response.  We have these generators producing large (larger than memory set) responses in a way that allows us to only keep small chunks in memory at a time.  However, it looks like the BaseHTTPMiddleware implementation uses an asyncio queue to store the individual chunks:
  
  https://github.com/encode/starlette/blob/master/starlette/middleware/base.py#L30
  
  This prevents any network backpressure handling -- if the client that is receiving the streaming response is on a slow connection, the queue will happily grow without bound and consume all memory, triggering kernel out-of-memory, when the ideal handling here would be for send to block (yield) when this happens.  I believe this would naturally happen if there were no queue here at all, so I am wondering why it needs to be here?
  
  Would a PR to remove the queueing be accepted?
  
  If not, what is the appropriate way to override this to not use a queue?  We can write our own, but the use of BaseHTTPMiddleware is hardcoded: https://github.com/encode/starlette/blob/519f5750b5e797bb3d4805fd29657674304ce397/starlette/applications.py#L197, leaving only some fairly hacky approaches to preventing this queueing.
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `starlette/middleware/base.py`
  Content:
  ```
  1 import asyncio
  2 import typing
  3 
  4 from starlette.requests import Request
  5 from starlette.responses import Response, StreamingResponse
  6 from starlette.types import ASGIApp, Receive, Scope, Send
  7 
  8 RequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]
  9 DispatchFunction = typing.Callable[
  10     [Request, RequestResponseEndpoint], typing.Awaitable[Response]
  11 ]
  12 
  13 
  14 class BaseHTTPMiddleware:
  15     def __init__(self, app: ASGIApp, dispatch: DispatchFunction = None) -> None:
  16         self.app = app
  17         self.dispatch_func = self.dispatch if dispatch is None else dispatch
  18 
  19     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
  20         if scope["type"] != "http":
  21             await self.app(scope, receive, send)
  22             return
  23 
  24         request = Request(scope, receive=receive)
  25         response = await self.dispatch_func(request, self.call_next)
  26         await response(scope, receive, send)
  27 
  28     async def call_next(self, request: Request) -> Response:
  29         loop = asyncio.get_event_loop()
  30         queue = asyncio.Queue()  # type: asyncio.Queue
  31 
  32         scope = request.scope
  33         receive = request.receive
  34         send = queue.put
  35 
  36         async def coro() -> None:
  37             try:
  38                 await self.app(scope, receive, send)
  39             finally:
  40                 await queue.put(None)
  41 
  42         task = loop.create_task(coro())
  43         message = await queue.get()
  44         if message is None:
  45             task.result()
  46             raise RuntimeError("No response returned.")
  47         assert message["type"] == "http.response.start"
  48 
  49         async def body_stream() -> typing.AsyncGenerator[bytes, None]:
  50             while True:
  51                 message = await queue.get()
  52                 if message is None:
  53                     break
  54                 assert message["type"] == "http.response.body"
  55                 yield message.get("body", b"")
  56             task.result()
  57 
  58         response = StreamingResponse(
  59             status_code=message["status"], content=body_stream()
  60         )
  61         response.raw_headers = message["headers"]
  62         return response
  63 
  64     async def dispatch(
  65         self, request: Request, call_next: RequestResponseEndpoint
  66     ) -> Response:
  67         raise NotImplementedError()  # pragma: no cover
  68
  ```
  --- END FILES ---
  
  Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.
  
  Here is an example:
  
  ```diff
  diff --git a/examples/server_async.py b/examples/server_async.py
  --- a/examples/server_async.py
  +++ b/examples/server_async.py
  @@ -313,4 +313,4 @@
   
   
   if __name__ == "__main__":
  -    asyncio.run(run_async_server("."), debug=True)
  +    asyncio.run(run_async_server(), debug=True)
  diff --git a/examples/server_sync.py b/examples/server_sync.py
  --- a/examples/server_sync.py
  +++ b/examples/server_sync.py
  @@ -313,5 +313,5 @@
   
   
   if __name__ == "__main__":
  -    server = run_sync_server(".")
  +    server = run_sync_server()
       server.shutdown()
  
  ```
  
golden_diff: diff --git a/starlette/middleware/base.py b/starlette/middleware/base.py
  --- a/starlette/middleware/base.py
  +++ b/starlette/middleware/base.py
  @@ -3,7 +3,7 @@
   
   from starlette.requests import Request
   from starlette.responses import Response, StreamingResponse
  -from starlette.types import ASGIApp, Receive, Scope, Send
  +from starlette.types import ASGIApp, Message, Receive, Scope, Send
   
   RequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]
   DispatchFunction = typing.Callable[
  @@ -27,7 +27,7 @@
   
       async def call_next(self, request: Request) -> Response:
           loop = asyncio.get_event_loop()
  -        queue = asyncio.Queue()  # type: asyncio.Queue
  +        queue: "asyncio.Queue[typing.Optional[Message]]" = asyncio.Queue(maxsize=1)
   
           scope = request.scope
           receive = request.receive
  
verification_info: {"golden_diff": "diff --git a/starlette/middleware/base.py b/starlette/middleware/base.py\n--- a/starlette/middleware/base.py\n+++ b/starlette/middleware/base.py\n@@ -3,7 +3,7 @@\n \n from starlette.requests import Request\n from starlette.responses import Response, StreamingResponse\n-from starlette.types import ASGIApp, Receive, Scope, Send\n+from starlette.types import ASGIApp, Message, Receive, Scope, Send\n \n RequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]\n DispatchFunction = typing.Callable[\n@@ -27,7 +27,7 @@\n \n     async def call_next(self, request: Request) -> Response:\n         loop = asyncio.get_event_loop()\n-        queue = asyncio.Queue()  # type: asyncio.Queue\n+        queue: \"asyncio.Queue[typing.Optional[Message]]\" = asyncio.Queue(maxsize=1)\n \n         scope = request.scope\n         receive = request.receive\n", "issue": "Memory usage streaming large responses\nWe've been running into memory issues when providing very large async generators to a streaming response.  We have these generators producing large (larger than memory set) responses in a way that allows us to only keep small chunks in memory at a time.  However, it looks like the BaseHTTPMiddleware implementation uses an asyncio queue to store the individual chunks:\r\n\r\nhttps://github.com/encode/starlette/blob/master/starlette/middleware/base.py#L30\r\n\r\nThis prevents any network backpressure handling -- if the client that is receiving the streaming response is on a slow connection, the queue will happily grow without bound and consume all memory, triggering kernel out-of-memory, when the ideal handling here would be for send to block (yield) when this happens.  I believe this would naturally happen if there were no queue here at all, so I am wondering why it needs to be here?\r\n\r\nWould a PR to remove the queueing be accepted?\r\n\r\nIf not, what is the appropriate way to override this to not use a queue?  We can write our own, but the use of BaseHTTPMiddleware is hardcoded: https://github.com/encode/starlette/blob/519f5750b5e797bb3d4805fd29657674304ce397/starlette/applications.py#L197, leaving only some fairly hacky approaches to preventing this queueing.\n", "before_files": [{"content": "import asyncio\nimport typing\n\nfrom starlette.requests import Request\nfrom starlette.responses import Response, StreamingResponse\nfrom starlette.types import ASGIApp, Receive, Scope, Send\n\nRequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]\nDispatchFunction = typing.Callable[\n    [Request, RequestResponseEndpoint], typing.Awaitable[Response]\n]\n\n\nclass BaseHTTPMiddleware:\n    def __init__(self, app: ASGIApp, dispatch: DispatchFunction = None) -> None:\n        self.app = app\n        self.dispatch_func = self.dispatch if dispatch is None else dispatch\n\n    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:\n        if scope[\"type\"] != \"http\":\n            await self.app(scope, receive, send)\n            return\n\n        request = Request(scope, receive=receive)\n        response = await self.dispatch_func(request, self.call_next)\n        await response(scope, receive, send)\n\n    async def call_next(self, request: Request) -> Response:\n        loop = asyncio.get_event_loop()\n        queue = asyncio.Queue()  # type: asyncio.Queue\n\n        scope = request.scope\n        receive = request.receive\n        send = queue.put\n\n        async def coro() -> None:\n            try:\n                await self.app(scope, receive, send)\n            finally:\n                await queue.put(None)\n\n        task = loop.create_task(coro())\n        message = await queue.get()\n        if message is None:\n            task.result()\n            raise RuntimeError(\"No response returned.\")\n        assert message[\"type\"] == \"http.response.start\"\n\n        async def body_stream() -> typing.AsyncGenerator[bytes, None]:\n            while True:\n                message = await queue.get()\n                if message is None:\n                    break\n                assert message[\"type\"] == \"http.response.body\"\n                yield message.get(\"body\", b\"\")\n            task.result()\n\n        response = StreamingResponse(\n            status_code=message[\"status\"], content=body_stream()\n        )\n        response.raw_headers = message[\"headers\"]\n        return response\n\n    async def dispatch(\n        self, request: Request, call_next: RequestResponseEndpoint\n    ) -> Response:\n        raise NotImplementedError()  # pragma: no cover\n", "path": "starlette/middleware/base.py"}], "after_files": [{"content": "import asyncio\nimport typing\n\nfrom starlette.requests import Request\nfrom starlette.responses import Response, StreamingResponse\nfrom starlette.types import ASGIApp, Message, Receive, Scope, Send\n\nRequestResponseEndpoint = typing.Callable[[Request], typing.Awaitable[Response]]\nDispatchFunction = typing.Callable[\n    [Request, RequestResponseEndpoint], typing.Awaitable[Response]\n]\n\n\nclass BaseHTTPMiddleware:\n    def __init__(self, app: ASGIApp, dispatch: DispatchFunction = None) -> None:\n        self.app = app\n        self.dispatch_func = self.dispatch if dispatch is None else dispatch\n\n    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:\n        if scope[\"type\"] != \"http\":\n            await self.app(scope, receive, send)\n            return\n\n        request = Request(scope, receive=receive)\n        response = await self.dispatch_func(request, self.call_next)\n        await response(scope, receive, send)\n\n    async def call_next(self, request: Request) -> Response:\n        loop = asyncio.get_event_loop()\n        queue: \"asyncio.Queue[typing.Optional[Message]]\" = asyncio.Queue(maxsize=1)\n\n        scope = request.scope\n        receive = request.receive\n        send = queue.put\n\n        async def coro() -> None:\n            try:\n                await self.app(scope, receive, send)\n            finally:\n                await queue.put(None)\n\n        task = loop.create_task(coro())\n        message = await queue.get()\n        if message is None:\n            task.result()\n            raise RuntimeError(\"No response returned.\")\n        assert message[\"type\"] == \"http.response.start\"\n\n        async def body_stream() -> typing.AsyncGenerator[bytes, None]:\n            while True:\n                message = await queue.get()\n                if message is None:\n                    break\n                assert message[\"type\"] == \"http.response.body\"\n                yield message.get(\"body\", b\"\")\n            task.result()\n\n        response = StreamingResponse(\n            status_code=message[\"status\"], content=body_stream()\n        )\n        response.raw_headers = message[\"headers\"]\n        return response\n\n    async def dispatch(\n        self, request: Request, call_next: RequestResponseEndpoint\n    ) -> Response:\n        raise NotImplementedError()  # pragma: no cover\n", "path": "starlette/middleware/base.py"}]}