We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Make ASGI dispatcher asyncio-agnostic
*Note: this issue breaks apart steps originally taken in #217.*

The ASGI dispatcher currently relies on asyncio-specific APIs to perform various operations, which is a blocker towards supporting alternative concurrency backends.

We want to make sure all of the I/O inside `ASGIDispatch` goes through the backend instead.

This will need the following, most of which can be tackled in independant PRs:

- [x] #257 Hard-code `self.backend = AsyncioBackend()` on `ASGIDispatch` to be able to iteratively defer asyncio-specific APIs to the backend.

- [x] #257 Introduce a queue-like interface:

```python
class BaseQueue:
    async def get(self): ...
    async def put(self, value: typing.Any): ...
```

along with a `.create_queue()` method on `ConcurrencyBackend`, to be used instead of/within the `BodyIterator`.

(For trio we'll probably need to add a `.close()` method as well, but that can be confirmed and tackled in due time.)

- [x] #260 Introduce a `BaseEvent` interface:

```python
class BaseEvent:
    def set(self): ...
    def is_set(self): ...
    async def wait(self): ...
```

along with a `.create_event()` method on `ConcurrencyBackend` to abstract `asyncio.Event()`.

- [x] #261 Remove usage of loop.create_task() via the background manager interface.

- [x] Lastly, turn the hard-coded `backend` into a parameter. Make sure it is optional and we use `AsyncioBackend` as a default.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `httpx/dispatch/asgi.py`
Content:
```
1 import typing
2 
3 from .base import AsyncDispatcher
4 from ..concurrency.base import ConcurrencyBackend
5 from ..concurrency.asyncio import AsyncioBackend
6 from ..config import CertTypes, TimeoutTypes, VerifyTypes
7 from ..models import AsyncRequest, AsyncResponse
8 
9 
10 class ASGIDispatch(AsyncDispatcher):
11     """
12     A custom dispatcher that handles sending requests directly to an ASGI app.
13 
14     The simplest way to use this functionality is to use the `app`argument.
15     This will automatically infer if 'app' is a WSGI or an ASGI application,
16     and will setup an appropriate dispatch class:
17 
18     ```
19     client = httpx.Client(app=app)
20     ```
21 
22     Alternatively, you can setup the dispatch instance explicitly.
23     This allows you to include any additional configuration arguments specific
24     to the ASGIDispatch class:
25 
26     ```
27     dispatch = httpx.ASGIDispatch(
28         app=app,
29         root_path="/submount",
30         client=("1.2.3.4", 123)
31     )
32     client = httpx.Client(dispatch=dispatch)
33 
34     Arguments:
35 
36     * `app` - The ASGI application.
37     * `raise_app_exceptions` - Boolean indicating if exceptions in the application
38        should be raised. Default to `True`. Can be set to `False` for use cases
39        such as testing the content of a client 500 response.
40     * `root_path` - The root path on which the ASGI application should be mounted.
41     * `client` - A two-tuple indicating the client IP and port of incoming requests.
42     ```
43     """
44 
45     def __init__(
46         self,
47         app: typing.Callable,
48         raise_app_exceptions: bool = True,
49         root_path: str = "",
50         client: typing.Tuple[str, int] = ("127.0.0.1", 123),
51     ) -> None:
52         self.app = app
53         self.raise_app_exceptions = raise_app_exceptions
54         self.root_path = root_path
55         self.client = client
56         # This will need to be turned into a parameter on this class at some point.
57         self.backend: ConcurrencyBackend = AsyncioBackend()
58 
59     async def send(
60         self,
61         request: AsyncRequest,
62         verify: VerifyTypes = None,
63         cert: CertTypes = None,
64         timeout: TimeoutTypes = None,
65     ) -> AsyncResponse:
66 
67         scope = {
68             "type": "http",
69             "asgi": {"version": "3.0"},
70             "http_version": "1.1",
71             "method": request.method,
72             "headers": request.headers.raw,
73             "scheme": request.url.scheme,
74             "path": request.url.path,
75             "query_string": request.url.query.encode("ascii"),
76             "server": request.url.host,
77             "client": self.client,
78             "root_path": self.root_path,
79         }
80         app = self.app
81         app_exc = None
82         status_code = None
83         headers = None
84         response_started_or_failed = self.backend.create_event()
85         response_body = BodyIterator(self.backend)
86         request_stream = request.stream()
87 
88         async def receive() -> dict:
89             nonlocal request_stream
90 
91             try:
92                 body = await request_stream.__anext__()
93             except StopAsyncIteration:
94                 return {"type": "http.request", "body": b"", "more_body": False}
95             return {"type": "http.request", "body": body, "more_body": True}
96 
97         async def send(message: dict) -> None:
98             nonlocal status_code, headers, response_started_or_failed
99             nonlocal response_body, request
100 
101             if message["type"] == "http.response.start":
102                 status_code = message["status"]
103                 headers = message.get("headers", [])
104                 response_started_or_failed.set()
105             elif message["type"] == "http.response.body":
106                 body = message.get("body", b"")
107                 more_body = message.get("more_body", False)
108                 if body and request.method != "HEAD":
109                     await response_body.put(body)
110                 if not more_body:
111                     await response_body.mark_as_done()
112 
113         async def run_app() -> None:
114             nonlocal app, scope, receive, send, app_exc, response_body
115             try:
116                 await app(scope, receive, send)
117             except Exception as exc:
118                 app_exc = exc
119             finally:
120                 await response_body.mark_as_done()
121                 response_started_or_failed.set()
122 
123         # Using the background manager here *works*, but it is weak design because
124         # the background task isn't strictly context-managed.
125         # We could consider refactoring the other uses of this abstraction
126         # (mainly sending/receiving request/response data in h11 and h2 dispatchers),
127         # and see if that allows us to come back here and refactor things out.
128         background = await self.backend.background_manager(run_app).__aenter__()
129 
130         await response_started_or_failed.wait()
131 
132         if app_exc is not None and self.raise_app_exceptions:
133             raise app_exc
134 
135         assert status_code is not None, "application did not return a response."
136         assert headers is not None
137 
138         async def on_close() -> None:
139             nonlocal response_body
140             await response_body.drain()
141             await background.__aexit__(None, None, None)
142             if app_exc is not None and self.raise_app_exceptions:
143                 raise app_exc
144 
145         return AsyncResponse(
146             status_code=status_code,
147             http_version="HTTP/1.1",
148             headers=headers,
149             content=response_body.iterate(),
150             on_close=on_close,
151             request=request,
152         )
153 
154 
155 class BodyIterator:
156     """
157     Provides a byte-iterator interface that the client can use to
158     ingest the response content from.
159     """
160 
161     def __init__(self, backend: ConcurrencyBackend) -> None:
162         self._queue = backend.create_queue(max_size=1)
163         self._done = object()
164 
165     async def iterate(self) -> typing.AsyncIterator[bytes]:
166         """
167         A byte-iterator, used by the client to consume the response body.
168         """
169         while True:
170             data = await self._queue.get()
171             if data is self._done:
172                 break
173             assert isinstance(data, bytes)
174             yield data
175 
176     async def drain(self) -> None:
177         """
178         Drain any remaining body, in order to allow any blocked `put()` calls
179         to complete.
180         """
181         async for chunk in self.iterate():
182             pass  # pragma: no cover
183 
184     async def put(self, data: bytes) -> None:
185         """
186         Used by the server to add data to the response body.
187         """
188         await self._queue.put(data)
189 
190     async def mark_as_done(self) -> None:
191         """
192         Used by the server to signal the end of the response body.
193         """
194         await self._queue.put(self._done)
195
```
--- 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/httpx/dispatch/asgi.py b/httpx/dispatch/asgi.py
--- a/httpx/dispatch/asgi.py
+++ b/httpx/dispatch/asgi.py
@@ -48,13 +48,13 @@
         raise_app_exceptions: bool = True,
         root_path: str = "",
         client: typing.Tuple[str, int] = ("127.0.0.1", 123),
+        backend: ConcurrencyBackend = None,
     ) -> None:
         self.app = app
         self.raise_app_exceptions = raise_app_exceptions
         self.root_path = root_path
         self.client = client
-        # This will need to be turned into a parameter on this class at some point.
-        self.backend: ConcurrencyBackend = AsyncioBackend()
+        self.backend = AsyncioBackend() if backend is None else backend
 
     async def send(
         self,

{"golden_diff": "diff --git a/httpx/dispatch/asgi.py b/httpx/dispatch/asgi.py\n--- a/httpx/dispatch/asgi.py\n+++ b/httpx/dispatch/asgi.py\n@@ -48,13 +48,13 @@\n         raise_app_exceptions: bool = True,\n         root_path: str = \"\",\n         client: typing.Tuple[str, int] = (\"127.0.0.1\", 123),\n+        backend: ConcurrencyBackend = None,\n     ) -> None:\n         self.app = app\n         self.raise_app_exceptions = raise_app_exceptions\n         self.root_path = root_path\n         self.client = client\n-        # This will need to be turned into a parameter on this class at some point.\n-        self.backend: ConcurrencyBackend = AsyncioBackend()\n+        self.backend = AsyncioBackend() if backend is None else backend\n \n     async def send(\n         self,\n", "issue": "Make ASGI dispatcher asyncio-agnostic\n*Note: this issue breaks apart steps originally taken in #217.*\r\n\r\nThe ASGI dispatcher currently relies on asyncio-specific APIs to perform various operations, which is a blocker towards supporting alternative concurrency backends.\r\n\r\nWe want to make sure all of the I/O inside `ASGIDispatch` goes through the backend instead.\r\n\r\nThis will need the following, most of which can be tackled in independant PRs:\r\n\r\n- [x] #257 Hard-code `self.backend = AsyncioBackend()` on `ASGIDispatch` to be able to iteratively defer asyncio-specific APIs to the backend.\r\n\r\n- [x] #257 Introduce a queue-like interface:\r\n\r\n```python\r\nclass BaseQueue:\r\n    async def get(self): ...\r\n    async def put(self, value: typing.Any): ...\r\n```\r\n\r\nalong with a `.create_queue()` method on `ConcurrencyBackend`, to be used instead of/within the `BodyIterator`.\r\n\r\n(For trio we'll probably need to add a `.close()` method as well, but that can be confirmed and tackled in due time.)\r\n\r\n- [x] #260 Introduce a `BaseEvent` interface:\r\n\r\n```python\r\nclass BaseEvent:\r\n    def set(self): ...\r\n    def is_set(self): ...\r\n    async def wait(self): ...\r\n```\r\n\r\nalong with a `.create_event()` method on `ConcurrencyBackend` to abstract `asyncio.Event()`.\r\n\r\n- [x] #261 Remove usage of loop.create_task() via the background manager interface.\r\n\r\n- [x] Lastly, turn the hard-coded `backend` into a parameter. Make sure it is optional and we use `AsyncioBackend` as a default.\n", "before_files": [{"content": "import typing\n\nfrom .base import AsyncDispatcher\nfrom ..concurrency.base import ConcurrencyBackend\nfrom ..concurrency.asyncio import AsyncioBackend\nfrom ..config import CertTypes, TimeoutTypes, VerifyTypes\nfrom ..models import AsyncRequest, AsyncResponse\n\n\nclass ASGIDispatch(AsyncDispatcher):\n    \"\"\"\n    A custom dispatcher that handles sending requests directly to an ASGI app.\n\n    The simplest way to use this functionality is to use the `app`argument.\n    This will automatically infer if 'app' is a WSGI or an ASGI application,\n    and will setup an appropriate dispatch class:\n\n    ```\n    client = httpx.Client(app=app)\n    ```\n\n    Alternatively, you can setup the dispatch instance explicitly.\n    This allows you to include any additional configuration arguments specific\n    to the ASGIDispatch class:\n\n    ```\n    dispatch = httpx.ASGIDispatch(\n        app=app,\n        root_path=\"/submount\",\n        client=(\"1.2.3.4\", 123)\n    )\n    client = httpx.Client(dispatch=dispatch)\n\n    Arguments:\n\n    * `app` - The ASGI application.\n    * `raise_app_exceptions` - Boolean indicating if exceptions in the application\n       should be raised. Default to `True`. Can be set to `False` for use cases\n       such as testing the content of a client 500 response.\n    * `root_path` - The root path on which the ASGI application should be mounted.\n    * `client` - A two-tuple indicating the client IP and port of incoming requests.\n    ```\n    \"\"\"\n\n    def __init__(\n        self,\n        app: typing.Callable,\n        raise_app_exceptions: bool = True,\n        root_path: str = \"\",\n        client: typing.Tuple[str, int] = (\"127.0.0.1\", 123),\n    ) -> None:\n        self.app = app\n        self.raise_app_exceptions = raise_app_exceptions\n        self.root_path = root_path\n        self.client = client\n        # This will need to be turned into a parameter on this class at some point.\n        self.backend: ConcurrencyBackend = AsyncioBackend()\n\n    async def send(\n        self,\n        request: AsyncRequest,\n        verify: VerifyTypes = None,\n        cert: CertTypes = None,\n        timeout: TimeoutTypes = None,\n    ) -> AsyncResponse:\n\n        scope = {\n            \"type\": \"http\",\n            \"asgi\": {\"version\": \"3.0\"},\n            \"http_version\": \"1.1\",\n            \"method\": request.method,\n            \"headers\": request.headers.raw,\n            \"scheme\": request.url.scheme,\n            \"path\": request.url.path,\n            \"query_string\": request.url.query.encode(\"ascii\"),\n            \"server\": request.url.host,\n            \"client\": self.client,\n            \"root_path\": self.root_path,\n        }\n        app = self.app\n        app_exc = None\n        status_code = None\n        headers = None\n        response_started_or_failed = self.backend.create_event()\n        response_body = BodyIterator(self.backend)\n        request_stream = request.stream()\n\n        async def receive() -> dict:\n            nonlocal request_stream\n\n            try:\n                body = await request_stream.__anext__()\n            except StopAsyncIteration:\n                return {\"type\": \"http.request\", \"body\": b\"\", \"more_body\": False}\n            return {\"type\": \"http.request\", \"body\": body, \"more_body\": True}\n\n        async def send(message: dict) -> None:\n            nonlocal status_code, headers, response_started_or_failed\n            nonlocal response_body, request\n\n            if message[\"type\"] == \"http.response.start\":\n                status_code = message[\"status\"]\n                headers = message.get(\"headers\", [])\n                response_started_or_failed.set()\n            elif message[\"type\"] == \"http.response.body\":\n                body = message.get(\"body\", b\"\")\n                more_body = message.get(\"more_body\", False)\n                if body and request.method != \"HEAD\":\n                    await response_body.put(body)\n                if not more_body:\n                    await response_body.mark_as_done()\n\n        async def run_app() -> None:\n            nonlocal app, scope, receive, send, app_exc, response_body\n            try:\n                await app(scope, receive, send)\n            except Exception as exc:\n                app_exc = exc\n            finally:\n                await response_body.mark_as_done()\n                response_started_or_failed.set()\n\n        # Using the background manager here *works*, but it is weak design because\n        # the background task isn't strictly context-managed.\n        # We could consider refactoring the other uses of this abstraction\n        # (mainly sending/receiving request/response data in h11 and h2 dispatchers),\n        # and see if that allows us to come back here and refactor things out.\n        background = await self.backend.background_manager(run_app).__aenter__()\n\n        await response_started_or_failed.wait()\n\n        if app_exc is not None and self.raise_app_exceptions:\n            raise app_exc\n\n        assert status_code is not None, \"application did not return a response.\"\n        assert headers is not None\n\n        async def on_close() -> None:\n            nonlocal response_body\n            await response_body.drain()\n            await background.__aexit__(None, None, None)\n            if app_exc is not None and self.raise_app_exceptions:\n                raise app_exc\n\n        return AsyncResponse(\n            status_code=status_code,\n            http_version=\"HTTP/1.1\",\n            headers=headers,\n            content=response_body.iterate(),\n            on_close=on_close,\n            request=request,\n        )\n\n\nclass BodyIterator:\n    \"\"\"\n    Provides a byte-iterator interface that the client can use to\n    ingest the response content from.\n    \"\"\"\n\n    def __init__(self, backend: ConcurrencyBackend) -> None:\n        self._queue = backend.create_queue(max_size=1)\n        self._done = object()\n\n    async def iterate(self) -> typing.AsyncIterator[bytes]:\n        \"\"\"\n        A byte-iterator, used by the client to consume the response body.\n        \"\"\"\n        while True:\n            data = await self._queue.get()\n            if data is self._done:\n                break\n            assert isinstance(data, bytes)\n            yield data\n\n    async def drain(self) -> None:\n        \"\"\"\n        Drain any remaining body, in order to allow any blocked `put()` calls\n        to complete.\n        \"\"\"\n        async for chunk in self.iterate():\n            pass  # pragma: no cover\n\n    async def put(self, data: bytes) -> None:\n        \"\"\"\n        Used by the server to add data to the response body.\n        \"\"\"\n        await self._queue.put(data)\n\n    async def mark_as_done(self) -> None:\n        \"\"\"\n        Used by the server to signal the end of the response body.\n        \"\"\"\n        await self._queue.put(self._done)\n", "path": "httpx/dispatch/asgi.py"}], "after_files": [{"content": "import typing\n\nfrom .base import AsyncDispatcher\nfrom ..concurrency.base import ConcurrencyBackend\nfrom ..concurrency.asyncio import AsyncioBackend\nfrom ..config import CertTypes, TimeoutTypes, VerifyTypes\nfrom ..models import AsyncRequest, AsyncResponse\n\n\nclass ASGIDispatch(AsyncDispatcher):\n    \"\"\"\n    A custom dispatcher that handles sending requests directly to an ASGI app.\n\n    The simplest way to use this functionality is to use the `app`argument.\n    This will automatically infer if 'app' is a WSGI or an ASGI application,\n    and will setup an appropriate dispatch class:\n\n    ```\n    client = httpx.Client(app=app)\n    ```\n\n    Alternatively, you can setup the dispatch instance explicitly.\n    This allows you to include any additional configuration arguments specific\n    to the ASGIDispatch class:\n\n    ```\n    dispatch = httpx.ASGIDispatch(\n        app=app,\n        root_path=\"/submount\",\n        client=(\"1.2.3.4\", 123)\n    )\n    client = httpx.Client(dispatch=dispatch)\n\n    Arguments:\n\n    * `app` - The ASGI application.\n    * `raise_app_exceptions` - Boolean indicating if exceptions in the application\n       should be raised. Default to `True`. Can be set to `False` for use cases\n       such as testing the content of a client 500 response.\n    * `root_path` - The root path on which the ASGI application should be mounted.\n    * `client` - A two-tuple indicating the client IP and port of incoming requests.\n    ```\n    \"\"\"\n\n    def __init__(\n        self,\n        app: typing.Callable,\n        raise_app_exceptions: bool = True,\n        root_path: str = \"\",\n        client: typing.Tuple[str, int] = (\"127.0.0.1\", 123),\n        backend: ConcurrencyBackend = None,\n    ) -> None:\n        self.app = app\n        self.raise_app_exceptions = raise_app_exceptions\n        self.root_path = root_path\n        self.client = client\n        self.backend = AsyncioBackend() if backend is None else backend\n\n    async def send(\n        self,\n        request: AsyncRequest,\n        verify: VerifyTypes = None,\n        cert: CertTypes = None,\n        timeout: TimeoutTypes = None,\n    ) -> AsyncResponse:\n\n        scope = {\n            \"type\": \"http\",\n            \"asgi\": {\"version\": \"3.0\"},\n            \"http_version\": \"1.1\",\n            \"method\": request.method,\n            \"headers\": request.headers.raw,\n            \"scheme\": request.url.scheme,\n            \"path\": request.url.path,\n            \"query_string\": request.url.query.encode(\"ascii\"),\n            \"server\": request.url.host,\n            \"client\": self.client,\n            \"root_path\": self.root_path,\n        }\n        app = self.app\n        app_exc = None\n        status_code = None\n        headers = None\n        response_started_or_failed = self.backend.create_event()\n        response_body = BodyIterator(self.backend)\n        request_stream = request.stream()\n\n        async def receive() -> dict:\n            nonlocal request_stream\n\n            try:\n                body = await request_stream.__anext__()\n            except StopAsyncIteration:\n                return {\"type\": \"http.request\", \"body\": b\"\", \"more_body\": False}\n            return {\"type\": \"http.request\", \"body\": body, \"more_body\": True}\n\n        async def send(message: dict) -> None:\n            nonlocal status_code, headers, response_started_or_failed\n            nonlocal response_body, request\n\n            if message[\"type\"] == \"http.response.start\":\n                status_code = message[\"status\"]\n                headers = message.get(\"headers\", [])\n                response_started_or_failed.set()\n            elif message[\"type\"] == \"http.response.body\":\n                body = message.get(\"body\", b\"\")\n                more_body = message.get(\"more_body\", False)\n                if body and request.method != \"HEAD\":\n                    await response_body.put(body)\n                if not more_body:\n                    await response_body.mark_as_done()\n\n        async def run_app() -> None:\n            nonlocal app, scope, receive, send, app_exc, response_body\n            try:\n                await app(scope, receive, send)\n            except Exception as exc:\n                app_exc = exc\n            finally:\n                await response_body.mark_as_done()\n                response_started_or_failed.set()\n\n        # Using the background manager here *works*, but it is weak design because\n        # the background task isn't strictly context-managed.\n        # We could consider refactoring the other uses of this abstraction\n        # (mainly sending/receiving request/response data in h11 and h2 dispatchers),\n        # and see if that allows us to come back here and refactor things out.\n        background = await self.backend.background_manager(run_app).__aenter__()\n\n        await response_started_or_failed.wait()\n\n        if app_exc is not None and self.raise_app_exceptions:\n            raise app_exc\n\n        assert status_code is not None, \"application did not return a response.\"\n        assert headers is not None\n\n        async def on_close() -> None:\n            nonlocal response_body\n            await response_body.drain()\n            await background.__aexit__(None, None, None)\n            if app_exc is not None and self.raise_app_exceptions:\n                raise app_exc\n\n        return AsyncResponse(\n            status_code=status_code,\n            http_version=\"HTTP/1.1\",\n            headers=headers,\n            content=response_body.iterate(),\n            on_close=on_close,\n            request=request,\n        )\n\n\nclass BodyIterator:\n    \"\"\"\n    Provides a byte-iterator interface that the client can use to\n    ingest the response content from.\n    \"\"\"\n\n    def __init__(self, backend: ConcurrencyBackend) -> None:\n        self._queue = backend.create_queue(max_size=1)\n        self._done = object()\n\n    async def iterate(self) -> typing.AsyncIterator[bytes]:\n        \"\"\"\n        A byte-iterator, used by the client to consume the response body.\n        \"\"\"\n        while True:\n            data = await self._queue.get()\n            if data is self._done:\n                break\n            assert isinstance(data, bytes)\n            yield data\n\n    async def drain(self) -> None:\n        \"\"\"\n        Drain any remaining body, in order to allow any blocked `put()` calls\n        to complete.\n        \"\"\"\n        async for chunk in self.iterate():\n            pass  # pragma: no cover\n\n    async def put(self, data: bytes) -> None:\n        \"\"\"\n        Used by the server to add data to the response body.\n        \"\"\"\n        await self._queue.put(data)\n\n    async def mark_as_done(self) -> None:\n        \"\"\"\n        Used by the server to signal the end of the response body.\n        \"\"\"\n        await self._queue.put(self._done)\n", "path": "httpx/dispatch/asgi.py"}]}