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()

```

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

{"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"}]}