We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
HTTPX claims Python 3.6+ support when asyncio backend TCPStream.start_tls explicitly does not support Python 3.6
As stated, HTTPX explicitly does not support Python 3.6 [here](https://github.com/encode/httpx/blob/master/httpx/concurrency/asyncio.py#L61) but declares Python 3.6 support [here](https://github.com/encode/httpx/blob/master/setup.py#L40). This has caused problems with users of a package of mine that is supposed to support Python 3.6.

--- 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/concurrency/asyncio.py`
Content:
```
1 import asyncio
2 import functools
3 import ssl
4 import sys
5 import typing
6 from types import TracebackType
7 
8 from ..config import PoolLimits, TimeoutConfig
9 from ..exceptions import ConnectTimeout, PoolTimeout, ReadTimeout, WriteTimeout
10 from .base import (
11     BaseBackgroundManager,
12     BaseEvent,
13     BasePoolSemaphore,
14     BaseQueue,
15     BaseSocketStream,
16     ConcurrencyBackend,
17     TimeoutFlag,
18 )
19 
20 SSL_MONKEY_PATCH_APPLIED = False
21 
22 
23 def ssl_monkey_patch() -> None:
24     """
25     Monkey-patch for https://bugs.python.org/issue36709
26 
27     This prevents console errors when outstanding HTTPS connections
28     still exist at the point of exiting.
29 
30     Clients which have been opened using a `with` block, or which have
31     had `close()` closed, will not exhibit this issue in the first place.
32     """
33     MonkeyPatch = asyncio.selector_events._SelectorSocketTransport  # type: ignore
34 
35     _write = MonkeyPatch.write
36 
37     def _fixed_write(self, data: bytes) -> None:  # type: ignore
38         if self._loop and not self._loop.is_closed():
39             _write(self, data)
40 
41     MonkeyPatch.write = _fixed_write
42 
43 
44 class SocketStream(BaseSocketStream):
45     def __init__(
46         self,
47         stream_reader: asyncio.StreamReader,
48         stream_writer: asyncio.StreamWriter,
49         timeout: TimeoutConfig,
50     ):
51         self.stream_reader = stream_reader
52         self.stream_writer = stream_writer
53         self.timeout = timeout
54 
55         self._inner: typing.Optional[SocketStream] = None
56 
57     async def start_tls(
58         self, hostname: str, ssl_context: ssl.SSLContext, timeout: TimeoutConfig
59     ) -> "SocketStream":
60         loop = asyncio.get_event_loop()
61         if not hasattr(loop, "start_tls"):  # pragma: no cover
62             raise NotImplementedError(
63                 "asyncio.AbstractEventLoop.start_tls() is only available in Python 3.7+"
64             )
65 
66         stream_reader = asyncio.StreamReader()
67         protocol = asyncio.StreamReaderProtocol(stream_reader)
68         transport = self.stream_writer.transport
69 
70         loop_start_tls = loop.start_tls  # type: ignore
71         transport = await asyncio.wait_for(
72             loop_start_tls(
73                 transport=transport,
74                 protocol=protocol,
75                 sslcontext=ssl_context,
76                 server_hostname=hostname,
77             ),
78             timeout=timeout.connect_timeout,
79         )
80 
81         stream_reader.set_transport(transport)
82         stream_writer = asyncio.StreamWriter(
83             transport=transport, protocol=protocol, reader=stream_reader, loop=loop
84         )
85 
86         ssl_stream = SocketStream(stream_reader, stream_writer, self.timeout)
87         # When we return a new SocketStream with new StreamReader/StreamWriter instances
88         # we need to keep references to the old StreamReader/StreamWriter so that they
89         # are not garbage collected and closed while we're still using them.
90         ssl_stream._inner = self
91         return ssl_stream
92 
93     def get_http_version(self) -> str:
94         ssl_object = self.stream_writer.get_extra_info("ssl_object")
95 
96         if ssl_object is None:
97             return "HTTP/1.1"
98 
99         ident = ssl_object.selected_alpn_protocol()
100         return "HTTP/2" if ident == "h2" else "HTTP/1.1"
101 
102     async def read(
103         self, n: int, timeout: TimeoutConfig = None, flag: TimeoutFlag = None
104     ) -> bytes:
105         if timeout is None:
106             timeout = self.timeout
107 
108         while True:
109             # Check our flag at the first possible moment, and use a fine
110             # grained retry loop if we're not yet in read-timeout mode.
111             should_raise = flag is None or flag.raise_on_read_timeout
112             read_timeout = timeout.read_timeout if should_raise else 0.01
113             try:
114                 data = await asyncio.wait_for(self.stream_reader.read(n), read_timeout)
115                 break
116             except asyncio.TimeoutError:
117                 if should_raise:
118                     raise ReadTimeout() from None
119                 # FIX(py3.6): yield control back to the event loop to give it a chance
120                 # to cancel `.read(n)` before we retry.
121                 # This prevents concurrent `.read()` calls, which asyncio
122                 # doesn't seem to allow on 3.6.
123                 # See: https://github.com/encode/httpx/issues/382
124                 await asyncio.sleep(0)
125 
126         return data
127 
128     def write_no_block(self, data: bytes) -> None:
129         self.stream_writer.write(data)  # pragma: nocover
130 
131     async def write(
132         self, data: bytes, timeout: TimeoutConfig = None, flag: TimeoutFlag = None
133     ) -> None:
134         if not data:
135             return
136 
137         if timeout is None:
138             timeout = self.timeout
139 
140         self.stream_writer.write(data)
141         while True:
142             try:
143                 await asyncio.wait_for(  # type: ignore
144                     self.stream_writer.drain(), timeout.write_timeout
145                 )
146                 break
147             except asyncio.TimeoutError:
148                 # We check our flag at the first possible moment, in order to
149                 # allow us to suppress write timeouts, if we've since
150                 # switched over to read-timeout mode.
151                 should_raise = flag is None or flag.raise_on_write_timeout
152                 if should_raise:
153                     raise WriteTimeout() from None
154 
155     def is_connection_dropped(self) -> bool:
156         # Counter-intuitively, what we really want to know here is whether the socket is
157         # *readable*, i.e. whether it would return immediately with empty bytes if we
158         # called `.recv()` on it, indicating that the other end has closed the socket.
159         # See: https://github.com/encode/httpx/pull/143#issuecomment-515181778
160         #
161         # As it turns out, asyncio checks for readability in the background
162         # (see: https://github.com/encode/httpx/pull/276#discussion_r322000402),
163         # so checking for EOF or readability here would yield the same result.
164         #
165         # At the cost of rigour, we check for EOF instead of readability because asyncio
166         # does not expose any public API to check for readability.
167         # (For a solution that uses private asyncio APIs, see:
168         # https://github.com/encode/httpx/pull/143#issuecomment-515202982)
169 
170         return self.stream_reader.at_eof()
171 
172     async def close(self) -> None:
173         self.stream_writer.close()
174         if sys.version_info >= (3, 7):
175             await self.stream_writer.wait_closed()
176 
177 
178 class PoolSemaphore(BasePoolSemaphore):
179     def __init__(self, pool_limits: PoolLimits):
180         self.pool_limits = pool_limits
181 
182     @property
183     def semaphore(self) -> typing.Optional[asyncio.BoundedSemaphore]:
184         if not hasattr(self, "_semaphore"):
185             max_connections = self.pool_limits.hard_limit
186             if max_connections is None:
187                 self._semaphore = None
188             else:
189                 self._semaphore = asyncio.BoundedSemaphore(value=max_connections)
190         return self._semaphore
191 
192     async def acquire(self) -> None:
193         if self.semaphore is None:
194             return
195 
196         timeout = self.pool_limits.pool_timeout
197         try:
198             await asyncio.wait_for(self.semaphore.acquire(), timeout)
199         except asyncio.TimeoutError:
200             raise PoolTimeout()
201 
202     def release(self) -> None:
203         if self.semaphore is None:
204             return
205 
206         self.semaphore.release()
207 
208 
209 class AsyncioBackend(ConcurrencyBackend):
210     def __init__(self) -> None:
211         global SSL_MONKEY_PATCH_APPLIED
212 
213         if not SSL_MONKEY_PATCH_APPLIED:
214             ssl_monkey_patch()
215         SSL_MONKEY_PATCH_APPLIED = True
216 
217     @property
218     def loop(self) -> asyncio.AbstractEventLoop:
219         if not hasattr(self, "_loop"):
220             try:
221                 self._loop = asyncio.get_event_loop()
222             except RuntimeError:
223                 self._loop = asyncio.new_event_loop()
224         return self._loop
225 
226     async def open_tcp_stream(
227         self,
228         hostname: str,
229         port: int,
230         ssl_context: typing.Optional[ssl.SSLContext],
231         timeout: TimeoutConfig,
232     ) -> SocketStream:
233         try:
234             stream_reader, stream_writer = await asyncio.wait_for(  # type: ignore
235                 asyncio.open_connection(hostname, port, ssl=ssl_context),
236                 timeout.connect_timeout,
237             )
238         except asyncio.TimeoutError:
239             raise ConnectTimeout()
240 
241         return SocketStream(
242             stream_reader=stream_reader, stream_writer=stream_writer, timeout=timeout
243         )
244 
245     async def run_in_threadpool(
246         self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
247     ) -> typing.Any:
248         if kwargs:
249             # loop.run_in_executor doesn't accept 'kwargs', so bind them in here
250             func = functools.partial(func, **kwargs)
251         return await self.loop.run_in_executor(None, func, *args)
252 
253     def run(
254         self, coroutine: typing.Callable, *args: typing.Any, **kwargs: typing.Any
255     ) -> typing.Any:
256         loop = self.loop
257         if loop.is_running():
258             self._loop = asyncio.new_event_loop()
259         try:
260             return self.loop.run_until_complete(coroutine(*args, **kwargs))
261         finally:
262             self._loop = loop
263 
264     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
265         return PoolSemaphore(limits)
266 
267     def create_queue(self, max_size: int) -> BaseQueue:
268         return typing.cast(BaseQueue, asyncio.Queue(maxsize=max_size))
269 
270     def create_event(self) -> BaseEvent:
271         return typing.cast(BaseEvent, asyncio.Event())
272 
273     def background_manager(
274         self, coroutine: typing.Callable, *args: typing.Any
275     ) -> "BackgroundManager":
276         return BackgroundManager(coroutine, args)
277 
278 
279 class BackgroundManager(BaseBackgroundManager):
280     def __init__(self, coroutine: typing.Callable, args: typing.Any) -> None:
281         self.coroutine = coroutine
282         self.args = args
283 
284     async def __aenter__(self) -> "BackgroundManager":
285         loop = asyncio.get_event_loop()
286         self.task = loop.create_task(self.coroutine(*self.args))
287         return self
288 
289     async def __aexit__(
290         self,
291         exc_type: typing.Type[BaseException] = None,
292         exc_value: BaseException = None,
293         traceback: TracebackType = None,
294     ) -> None:
295         await self.task
296         if exc_type is None:
297             self.task.result()
298
```
--- 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/concurrency/asyncio.py b/httpx/concurrency/asyncio.py
--- a/httpx/concurrency/asyncio.py
+++ b/httpx/concurrency/asyncio.py
@@ -58,16 +58,49 @@
         self, hostname: str, ssl_context: ssl.SSLContext, timeout: TimeoutConfig
     ) -> "SocketStream":
         loop = asyncio.get_event_loop()
-        if not hasattr(loop, "start_tls"):  # pragma: no cover
-            raise NotImplementedError(
-                "asyncio.AbstractEventLoop.start_tls() is only available in Python 3.7+"
-            )
 
         stream_reader = asyncio.StreamReader()
         protocol = asyncio.StreamReaderProtocol(stream_reader)
         transport = self.stream_writer.transport
 
-        loop_start_tls = loop.start_tls  # type: ignore
+        if hasattr(loop, "start_tls"):
+            loop_start_tls = loop.start_tls  # type: ignore
+        else:
+
+            async def loop_start_tls(
+                transport: asyncio.BaseTransport,
+                protocol: asyncio.BaseProtocol,
+                sslcontext: ssl.SSLContext = None,
+                *,
+                server_side: bool = False,
+                server_hostname: str = None,
+                ssl_handshake_timeout: float = None,
+            ) -> asyncio.Transport:
+                """Python 3.6 asyncio doesn't have a start_tls() method on the loop
+                so we use this function in place of the loop's start_tls() method.
+                Adapted from this comment:
+                https://github.com/urllib3/urllib3/issues/1323#issuecomment-362494839
+                """
+                import asyncio.sslproto
+
+                waiter = loop.create_future()
+                ssl_protocol = asyncio.sslproto.SSLProtocol(
+                    loop,
+                    protocol,
+                    sslcontext,
+                    waiter,
+                    server_side=False,
+                    server_hostname=server_hostname,
+                    call_connection_made=False,
+                )
+
+                transport.set_protocol(ssl_protocol)
+                loop.call_soon(ssl_protocol.connection_made, transport)
+                loop.call_soon(transport.resume_reading)  # type: ignore
+
+                await waiter
+                return ssl_protocol._app_transport
+
         transport = await asyncio.wait_for(
             loop_start_tls(
                 transport=transport,

{"golden_diff": "diff --git a/httpx/concurrency/asyncio.py b/httpx/concurrency/asyncio.py\n--- a/httpx/concurrency/asyncio.py\n+++ b/httpx/concurrency/asyncio.py\n@@ -58,16 +58,49 @@\n         self, hostname: str, ssl_context: ssl.SSLContext, timeout: TimeoutConfig\n     ) -> \"SocketStream\":\n         loop = asyncio.get_event_loop()\n-        if not hasattr(loop, \"start_tls\"):  # pragma: no cover\n-            raise NotImplementedError(\n-                \"asyncio.AbstractEventLoop.start_tls() is only available in Python 3.7+\"\n-            )\n \n         stream_reader = asyncio.StreamReader()\n         protocol = asyncio.StreamReaderProtocol(stream_reader)\n         transport = self.stream_writer.transport\n \n-        loop_start_tls = loop.start_tls  # type: ignore\n+        if hasattr(loop, \"start_tls\"):\n+            loop_start_tls = loop.start_tls  # type: ignore\n+        else:\n+\n+            async def loop_start_tls(\n+                transport: asyncio.BaseTransport,\n+                protocol: asyncio.BaseProtocol,\n+                sslcontext: ssl.SSLContext = None,\n+                *,\n+                server_side: bool = False,\n+                server_hostname: str = None,\n+                ssl_handshake_timeout: float = None,\n+            ) -> asyncio.Transport:\n+                \"\"\"Python 3.6 asyncio doesn't have a start_tls() method on the loop\n+                so we use this function in place of the loop's start_tls() method.\n+                Adapted from this comment:\n+                https://github.com/urllib3/urllib3/issues/1323#issuecomment-362494839\n+                \"\"\"\n+                import asyncio.sslproto\n+\n+                waiter = loop.create_future()\n+                ssl_protocol = asyncio.sslproto.SSLProtocol(\n+                    loop,\n+                    protocol,\n+                    sslcontext,\n+                    waiter,\n+                    server_side=False,\n+                    server_hostname=server_hostname,\n+                    call_connection_made=False,\n+                )\n+\n+                transport.set_protocol(ssl_protocol)\n+                loop.call_soon(ssl_protocol.connection_made, transport)\n+                loop.call_soon(transport.resume_reading)  # type: ignore\n+\n+                await waiter\n+                return ssl_protocol._app_transport\n+\n         transport = await asyncio.wait_for(\n             loop_start_tls(\n                 transport=transport,\n", "issue": "HTTPX claims Python 3.6+ support when asyncio backend TCPStream.start_tls explicitly does not support Python 3.6\nAs stated, HTTPX explicitly does not support Python 3.6 [here](https://github.com/encode/httpx/blob/master/httpx/concurrency/asyncio.py#L61) but declares Python 3.6 support [here](https://github.com/encode/httpx/blob/master/setup.py#L40). This has caused problems with users of a package of mine that is supposed to support Python 3.6.\n", "before_files": [{"content": "import asyncio\nimport functools\nimport ssl\nimport sys\nimport typing\nfrom types import TracebackType\n\nfrom ..config import PoolLimits, TimeoutConfig\nfrom ..exceptions import ConnectTimeout, PoolTimeout, ReadTimeout, WriteTimeout\nfrom .base import (\n    BaseBackgroundManager,\n    BaseEvent,\n    BasePoolSemaphore,\n    BaseQueue,\n    BaseSocketStream,\n    ConcurrencyBackend,\n    TimeoutFlag,\n)\n\nSSL_MONKEY_PATCH_APPLIED = False\n\n\ndef ssl_monkey_patch() -> None:\n    \"\"\"\n    Monkey-patch for https://bugs.python.org/issue36709\n\n    This prevents console errors when outstanding HTTPS connections\n    still exist at the point of exiting.\n\n    Clients which have been opened using a `with` block, or which have\n    had `close()` closed, will not exhibit this issue in the first place.\n    \"\"\"\n    MonkeyPatch = asyncio.selector_events._SelectorSocketTransport  # type: ignore\n\n    _write = MonkeyPatch.write\n\n    def _fixed_write(self, data: bytes) -> None:  # type: ignore\n        if self._loop and not self._loop.is_closed():\n            _write(self, data)\n\n    MonkeyPatch.write = _fixed_write\n\n\nclass SocketStream(BaseSocketStream):\n    def __init__(\n        self,\n        stream_reader: asyncio.StreamReader,\n        stream_writer: asyncio.StreamWriter,\n        timeout: TimeoutConfig,\n    ):\n        self.stream_reader = stream_reader\n        self.stream_writer = stream_writer\n        self.timeout = timeout\n\n        self._inner: typing.Optional[SocketStream] = None\n\n    async def start_tls(\n        self, hostname: str, ssl_context: ssl.SSLContext, timeout: TimeoutConfig\n    ) -> \"SocketStream\":\n        loop = asyncio.get_event_loop()\n        if not hasattr(loop, \"start_tls\"):  # pragma: no cover\n            raise NotImplementedError(\n                \"asyncio.AbstractEventLoop.start_tls() is only available in Python 3.7+\"\n            )\n\n        stream_reader = asyncio.StreamReader()\n        protocol = asyncio.StreamReaderProtocol(stream_reader)\n        transport = self.stream_writer.transport\n\n        loop_start_tls = loop.start_tls  # type: ignore\n        transport = await asyncio.wait_for(\n            loop_start_tls(\n                transport=transport,\n                protocol=protocol,\n                sslcontext=ssl_context,\n                server_hostname=hostname,\n            ),\n            timeout=timeout.connect_timeout,\n        )\n\n        stream_reader.set_transport(transport)\n        stream_writer = asyncio.StreamWriter(\n            transport=transport, protocol=protocol, reader=stream_reader, loop=loop\n        )\n\n        ssl_stream = SocketStream(stream_reader, stream_writer, self.timeout)\n        # When we return a new SocketStream with new StreamReader/StreamWriter instances\n        # we need to keep references to the old StreamReader/StreamWriter so that they\n        # are not garbage collected and closed while we're still using them.\n        ssl_stream._inner = self\n        return ssl_stream\n\n    def get_http_version(self) -> str:\n        ssl_object = self.stream_writer.get_extra_info(\"ssl_object\")\n\n        if ssl_object is None:\n            return \"HTTP/1.1\"\n\n        ident = ssl_object.selected_alpn_protocol()\n        return \"HTTP/2\" if ident == \"h2\" else \"HTTP/1.1\"\n\n    async def read(\n        self, n: int, timeout: TimeoutConfig = None, flag: TimeoutFlag = None\n    ) -> bytes:\n        if timeout is None:\n            timeout = self.timeout\n\n        while True:\n            # Check our flag at the first possible moment, and use a fine\n            # grained retry loop if we're not yet in read-timeout mode.\n            should_raise = flag is None or flag.raise_on_read_timeout\n            read_timeout = timeout.read_timeout if should_raise else 0.01\n            try:\n                data = await asyncio.wait_for(self.stream_reader.read(n), read_timeout)\n                break\n            except asyncio.TimeoutError:\n                if should_raise:\n                    raise ReadTimeout() from None\n                # FIX(py3.6): yield control back to the event loop to give it a chance\n                # to cancel `.read(n)` before we retry.\n                # This prevents concurrent `.read()` calls, which asyncio\n                # doesn't seem to allow on 3.6.\n                # See: https://github.com/encode/httpx/issues/382\n                await asyncio.sleep(0)\n\n        return data\n\n    def write_no_block(self, data: bytes) -> None:\n        self.stream_writer.write(data)  # pragma: nocover\n\n    async def write(\n        self, data: bytes, timeout: TimeoutConfig = None, flag: TimeoutFlag = None\n    ) -> None:\n        if not data:\n            return\n\n        if timeout is None:\n            timeout = self.timeout\n\n        self.stream_writer.write(data)\n        while True:\n            try:\n                await asyncio.wait_for(  # type: ignore\n                    self.stream_writer.drain(), timeout.write_timeout\n                )\n                break\n            except asyncio.TimeoutError:\n                # We check our flag at the first possible moment, in order to\n                # allow us to suppress write timeouts, if we've since\n                # switched over to read-timeout mode.\n                should_raise = flag is None or flag.raise_on_write_timeout\n                if should_raise:\n                    raise WriteTimeout() from None\n\n    def is_connection_dropped(self) -> bool:\n        # Counter-intuitively, what we really want to know here is whether the socket is\n        # *readable*, i.e. whether it would return immediately with empty bytes if we\n        # called `.recv()` on it, indicating that the other end has closed the socket.\n        # See: https://github.com/encode/httpx/pull/143#issuecomment-515181778\n        #\n        # As it turns out, asyncio checks for readability in the background\n        # (see: https://github.com/encode/httpx/pull/276#discussion_r322000402),\n        # so checking for EOF or readability here would yield the same result.\n        #\n        # At the cost of rigour, we check for EOF instead of readability because asyncio\n        # does not expose any public API to check for readability.\n        # (For a solution that uses private asyncio APIs, see:\n        # https://github.com/encode/httpx/pull/143#issuecomment-515202982)\n\n        return self.stream_reader.at_eof()\n\n    async def close(self) -> None:\n        self.stream_writer.close()\n        if sys.version_info >= (3, 7):\n            await self.stream_writer.wait_closed()\n\n\nclass PoolSemaphore(BasePoolSemaphore):\n    def __init__(self, pool_limits: PoolLimits):\n        self.pool_limits = pool_limits\n\n    @property\n    def semaphore(self) -> typing.Optional[asyncio.BoundedSemaphore]:\n        if not hasattr(self, \"_semaphore\"):\n            max_connections = self.pool_limits.hard_limit\n            if max_connections is None:\n                self._semaphore = None\n            else:\n                self._semaphore = asyncio.BoundedSemaphore(value=max_connections)\n        return self._semaphore\n\n    async def acquire(self) -> None:\n        if self.semaphore is None:\n            return\n\n        timeout = self.pool_limits.pool_timeout\n        try:\n            await asyncio.wait_for(self.semaphore.acquire(), timeout)\n        except asyncio.TimeoutError:\n            raise PoolTimeout()\n\n    def release(self) -> None:\n        if self.semaphore is None:\n            return\n\n        self.semaphore.release()\n\n\nclass AsyncioBackend(ConcurrencyBackend):\n    def __init__(self) -> None:\n        global SSL_MONKEY_PATCH_APPLIED\n\n        if not SSL_MONKEY_PATCH_APPLIED:\n            ssl_monkey_patch()\n        SSL_MONKEY_PATCH_APPLIED = True\n\n    @property\n    def loop(self) -> asyncio.AbstractEventLoop:\n        if not hasattr(self, \"_loop\"):\n            try:\n                self._loop = asyncio.get_event_loop()\n            except RuntimeError:\n                self._loop = asyncio.new_event_loop()\n        return self._loop\n\n    async def open_tcp_stream(\n        self,\n        hostname: str,\n        port: int,\n        ssl_context: typing.Optional[ssl.SSLContext],\n        timeout: TimeoutConfig,\n    ) -> SocketStream:\n        try:\n            stream_reader, stream_writer = await asyncio.wait_for(  # type: ignore\n                asyncio.open_connection(hostname, port, ssl=ssl_context),\n                timeout.connect_timeout,\n            )\n        except asyncio.TimeoutError:\n            raise ConnectTimeout()\n\n        return SocketStream(\n            stream_reader=stream_reader, stream_writer=stream_writer, timeout=timeout\n        )\n\n    async def run_in_threadpool(\n        self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any\n    ) -> typing.Any:\n        if kwargs:\n            # loop.run_in_executor doesn't accept 'kwargs', so bind them in here\n            func = functools.partial(func, **kwargs)\n        return await self.loop.run_in_executor(None, func, *args)\n\n    def run(\n        self, coroutine: typing.Callable, *args: typing.Any, **kwargs: typing.Any\n    ) -> typing.Any:\n        loop = self.loop\n        if loop.is_running():\n            self._loop = asyncio.new_event_loop()\n        try:\n            return self.loop.run_until_complete(coroutine(*args, **kwargs))\n        finally:\n            self._loop = loop\n\n    def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:\n        return PoolSemaphore(limits)\n\n    def create_queue(self, max_size: int) -> BaseQueue:\n        return typing.cast(BaseQueue, asyncio.Queue(maxsize=max_size))\n\n    def create_event(self) -> BaseEvent:\n        return typing.cast(BaseEvent, asyncio.Event())\n\n    def background_manager(\n        self, coroutine: typing.Callable, *args: typing.Any\n    ) -> \"BackgroundManager\":\n        return BackgroundManager(coroutine, args)\n\n\nclass BackgroundManager(BaseBackgroundManager):\n    def __init__(self, coroutine: typing.Callable, args: typing.Any) -> None:\n        self.coroutine = coroutine\n        self.args = args\n\n    async def __aenter__(self) -> \"BackgroundManager\":\n        loop = asyncio.get_event_loop()\n        self.task = loop.create_task(self.coroutine(*self.args))\n        return self\n\n    async def __aexit__(\n        self,\n        exc_type: typing.Type[BaseException] = None,\n        exc_value: BaseException = None,\n        traceback: TracebackType = None,\n    ) -> None:\n        await self.task\n        if exc_type is None:\n            self.task.result()\n", "path": "httpx/concurrency/asyncio.py"}], "after_files": [{"content": "import asyncio\nimport functools\nimport ssl\nimport sys\nimport typing\nfrom types import TracebackType\n\nfrom ..config import PoolLimits, TimeoutConfig\nfrom ..exceptions import ConnectTimeout, PoolTimeout, ReadTimeout, WriteTimeout\nfrom .base import (\n    BaseBackgroundManager,\n    BaseEvent,\n    BasePoolSemaphore,\n    BaseQueue,\n    BaseSocketStream,\n    ConcurrencyBackend,\n    TimeoutFlag,\n)\n\nSSL_MONKEY_PATCH_APPLIED = False\n\n\ndef ssl_monkey_patch() -> None:\n    \"\"\"\n    Monkey-patch for https://bugs.python.org/issue36709\n\n    This prevents console errors when outstanding HTTPS connections\n    still exist at the point of exiting.\n\n    Clients which have been opened using a `with` block, or which have\n    had `close()` closed, will not exhibit this issue in the first place.\n    \"\"\"\n    MonkeyPatch = asyncio.selector_events._SelectorSocketTransport  # type: ignore\n\n    _write = MonkeyPatch.write\n\n    def _fixed_write(self, data: bytes) -> None:  # type: ignore\n        if self._loop and not self._loop.is_closed():\n            _write(self, data)\n\n    MonkeyPatch.write = _fixed_write\n\n\nclass SocketStream(BaseSocketStream):\n    def __init__(\n        self,\n        stream_reader: asyncio.StreamReader,\n        stream_writer: asyncio.StreamWriter,\n        timeout: TimeoutConfig,\n    ):\n        self.stream_reader = stream_reader\n        self.stream_writer = stream_writer\n        self.timeout = timeout\n\n        self._inner: typing.Optional[SocketStream] = None\n\n    async def start_tls(\n        self, hostname: str, ssl_context: ssl.SSLContext, timeout: TimeoutConfig\n    ) -> \"SocketStream\":\n        loop = asyncio.get_event_loop()\n\n        stream_reader = asyncio.StreamReader()\n        protocol = asyncio.StreamReaderProtocol(stream_reader)\n        transport = self.stream_writer.transport\n\n        if hasattr(loop, \"start_tls\"):\n            loop_start_tls = loop.start_tls  # type: ignore\n        else:\n\n            async def loop_start_tls(\n                transport: asyncio.BaseTransport,\n                protocol: asyncio.BaseProtocol,\n                sslcontext: ssl.SSLContext = None,\n                *,\n                server_side: bool = False,\n                server_hostname: str = None,\n                ssl_handshake_timeout: float = None,\n            ) -> asyncio.Transport:\n                \"\"\"Python 3.6 asyncio doesn't have a start_tls() method on the loop\n                so we use this function in place of the loop's start_tls() method.\n                Adapted from this comment:\n                https://github.com/urllib3/urllib3/issues/1323#issuecomment-362494839\n                \"\"\"\n                import asyncio.sslproto\n\n                waiter = loop.create_future()\n                ssl_protocol = asyncio.sslproto.SSLProtocol(\n                    loop,\n                    protocol,\n                    sslcontext,\n                    waiter,\n                    server_side=False,\n                    server_hostname=server_hostname,\n                    call_connection_made=False,\n                )\n\n                transport.set_protocol(ssl_protocol)\n                loop.call_soon(ssl_protocol.connection_made, transport)\n                loop.call_soon(transport.resume_reading)  # type: ignore\n\n                await waiter\n                return ssl_protocol._app_transport\n\n        transport = await asyncio.wait_for(\n            loop_start_tls(\n                transport=transport,\n                protocol=protocol,\n                sslcontext=ssl_context,\n                server_hostname=hostname,\n            ),\n            timeout=timeout.connect_timeout,\n        )\n\n        stream_reader.set_transport(transport)\n        stream_writer = asyncio.StreamWriter(\n            transport=transport, protocol=protocol, reader=stream_reader, loop=loop\n        )\n\n        ssl_stream = SocketStream(stream_reader, stream_writer, self.timeout)\n        # When we return a new SocketStream with new StreamReader/StreamWriter instances\n        # we need to keep references to the old StreamReader/StreamWriter so that they\n        # are not garbage collected and closed while we're still using them.\n        ssl_stream._inner = self\n        return ssl_stream\n\n    def get_http_version(self) -> str:\n        ssl_object = self.stream_writer.get_extra_info(\"ssl_object\")\n\n        if ssl_object is None:\n            return \"HTTP/1.1\"\n\n        ident = ssl_object.selected_alpn_protocol()\n        return \"HTTP/2\" if ident == \"h2\" else \"HTTP/1.1\"\n\n    async def read(\n        self, n: int, timeout: TimeoutConfig = None, flag: TimeoutFlag = None\n    ) -> bytes:\n        if timeout is None:\n            timeout = self.timeout\n\n        while True:\n            # Check our flag at the first possible moment, and use a fine\n            # grained retry loop if we're not yet in read-timeout mode.\n            should_raise = flag is None or flag.raise_on_read_timeout\n            read_timeout = timeout.read_timeout if should_raise else 0.01\n            try:\n                data = await asyncio.wait_for(self.stream_reader.read(n), read_timeout)\n                break\n            except asyncio.TimeoutError:\n                if should_raise:\n                    raise ReadTimeout() from None\n                # FIX(py3.6): yield control back to the event loop to give it a chance\n                # to cancel `.read(n)` before we retry.\n                # This prevents concurrent `.read()` calls, which asyncio\n                # doesn't seem to allow on 3.6.\n                # See: https://github.com/encode/httpx/issues/382\n                await asyncio.sleep(0)\n\n        return data\n\n    def write_no_block(self, data: bytes) -> None:\n        self.stream_writer.write(data)  # pragma: nocover\n\n    async def write(\n        self, data: bytes, timeout: TimeoutConfig = None, flag: TimeoutFlag = None\n    ) -> None:\n        if not data:\n            return\n\n        if timeout is None:\n            timeout = self.timeout\n\n        self.stream_writer.write(data)\n        while True:\n            try:\n                await asyncio.wait_for(  # type: ignore\n                    self.stream_writer.drain(), timeout.write_timeout\n                )\n                break\n            except asyncio.TimeoutError:\n                # We check our flag at the first possible moment, in order to\n                # allow us to suppress write timeouts, if we've since\n                # switched over to read-timeout mode.\n                should_raise = flag is None or flag.raise_on_write_timeout\n                if should_raise:\n                    raise WriteTimeout() from None\n\n    def is_connection_dropped(self) -> bool:\n        # Counter-intuitively, what we really want to know here is whether the socket is\n        # *readable*, i.e. whether it would return immediately with empty bytes if we\n        # called `.recv()` on it, indicating that the other end has closed the socket.\n        # See: https://github.com/encode/httpx/pull/143#issuecomment-515181778\n        #\n        # As it turns out, asyncio checks for readability in the background\n        # (see: https://github.com/encode/httpx/pull/276#discussion_r322000402),\n        # so checking for EOF or readability here would yield the same result.\n        #\n        # At the cost of rigour, we check for EOF instead of readability because asyncio\n        # does not expose any public API to check for readability.\n        # (For a solution that uses private asyncio APIs, see:\n        # https://github.com/encode/httpx/pull/143#issuecomment-515202982)\n\n        return self.stream_reader.at_eof()\n\n    async def close(self) -> None:\n        self.stream_writer.close()\n        if sys.version_info >= (3, 7):\n            await self.stream_writer.wait_closed()\n\n\nclass PoolSemaphore(BasePoolSemaphore):\n    def __init__(self, pool_limits: PoolLimits):\n        self.pool_limits = pool_limits\n\n    @property\n    def semaphore(self) -> typing.Optional[asyncio.BoundedSemaphore]:\n        if not hasattr(self, \"_semaphore\"):\n            max_connections = self.pool_limits.hard_limit\n            if max_connections is None:\n                self._semaphore = None\n            else:\n                self._semaphore = asyncio.BoundedSemaphore(value=max_connections)\n        return self._semaphore\n\n    async def acquire(self) -> None:\n        if self.semaphore is None:\n            return\n\n        timeout = self.pool_limits.pool_timeout\n        try:\n            await asyncio.wait_for(self.semaphore.acquire(), timeout)\n        except asyncio.TimeoutError:\n            raise PoolTimeout()\n\n    def release(self) -> None:\n        if self.semaphore is None:\n            return\n\n        self.semaphore.release()\n\n\nclass AsyncioBackend(ConcurrencyBackend):\n    def __init__(self) -> None:\n        global SSL_MONKEY_PATCH_APPLIED\n\n        if not SSL_MONKEY_PATCH_APPLIED:\n            ssl_monkey_patch()\n        SSL_MONKEY_PATCH_APPLIED = True\n\n    @property\n    def loop(self) -> asyncio.AbstractEventLoop:\n        if not hasattr(self, \"_loop\"):\n            try:\n                self._loop = asyncio.get_event_loop()\n            except RuntimeError:\n                self._loop = asyncio.new_event_loop()\n        return self._loop\n\n    async def open_tcp_stream(\n        self,\n        hostname: str,\n        port: int,\n        ssl_context: typing.Optional[ssl.SSLContext],\n        timeout: TimeoutConfig,\n    ) -> SocketStream:\n        try:\n            stream_reader, stream_writer = await asyncio.wait_for(  # type: ignore\n                asyncio.open_connection(hostname, port, ssl=ssl_context),\n                timeout.connect_timeout,\n            )\n        except asyncio.TimeoutError:\n            raise ConnectTimeout()\n\n        return SocketStream(\n            stream_reader=stream_reader, stream_writer=stream_writer, timeout=timeout\n        )\n\n    async def run_in_threadpool(\n        self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any\n    ) -> typing.Any:\n        if kwargs:\n            # loop.run_in_executor doesn't accept 'kwargs', so bind them in here\n            func = functools.partial(func, **kwargs)\n        return await self.loop.run_in_executor(None, func, *args)\n\n    def run(\n        self, coroutine: typing.Callable, *args: typing.Any, **kwargs: typing.Any\n    ) -> typing.Any:\n        loop = self.loop\n        if loop.is_running():\n            self._loop = asyncio.new_event_loop()\n        try:\n            return self.loop.run_until_complete(coroutine(*args, **kwargs))\n        finally:\n            self._loop = loop\n\n    def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:\n        return PoolSemaphore(limits)\n\n    def create_queue(self, max_size: int) -> BaseQueue:\n        return typing.cast(BaseQueue, asyncio.Queue(maxsize=max_size))\n\n    def create_event(self) -> BaseEvent:\n        return typing.cast(BaseEvent, asyncio.Event())\n\n    def background_manager(\n        self, coroutine: typing.Callable, *args: typing.Any\n    ) -> \"BackgroundManager\":\n        return BackgroundManager(coroutine, args)\n\n\nclass BackgroundManager(BaseBackgroundManager):\n    def __init__(self, coroutine: typing.Callable, args: typing.Any) -> None:\n        self.coroutine = coroutine\n        self.args = args\n\n    async def __aenter__(self) -> \"BackgroundManager\":\n        loop = asyncio.get_event_loop()\n        self.task = loop.create_task(self.coroutine(*self.args))\n        return self\n\n    async def __aexit__(\n        self,\n        exc_type: typing.Type[BaseException] = None,\n        exc_value: BaseException = None,\n        traceback: TracebackType = None,\n    ) -> None:\n        await self.task\n        if exc_type is None:\n            self.task.result()\n", "path": "httpx/concurrency/asyncio.py"}]}