We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Cannot connect to yahoo.com, receiving a ProtocolError on Stream 0 after connection
Couple of issues here: we're not raising an exception after receiving a `ProtocolError` from HTTP/2 when connecting to `yahoo.com`. We should be bubbling that exception up to users.

The second issue is is we don't know why yahoo is giving us this error, we have to figure that out as I'm sure it'll impact more than just them.

I found this by running interop tests against Alexa 1000 websites.

--- 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/http2.py`
Content:
```
1 import typing
2 
3 import h2.connection
4 import h2.events
5 from h2.config import H2Configuration
6 from h2.settings import SettingCodes, Settings
7 
8 from ..backends.base import (
9     BaseEvent,
10     BaseSocketStream,
11     ConcurrencyBackend,
12     lookup_backend,
13 )
14 from ..config import Timeout
15 from ..content_streams import AsyncIteratorStream
16 from ..exceptions import ProtocolError
17 from ..models import Request, Response
18 from ..utils import get_logger
19 from .base import OpenConnection
20 
21 logger = get_logger(__name__)
22 
23 
24 class HTTP2Connection(OpenConnection):
25     READ_NUM_BYTES = 4096
26     CONFIG = H2Configuration(validate_inbound_headers=False)
27 
28     def __init__(
29         self,
30         socket: BaseSocketStream,
31         backend: typing.Union[str, ConcurrencyBackend] = "auto",
32         on_release: typing.Callable = None,
33     ):
34         self.socket = socket
35         self.backend = lookup_backend(backend)
36         self.on_release = on_release
37         self.state = h2.connection.H2Connection(config=self.CONFIG)
38 
39         self.streams = {}  # type: typing.Dict[int, HTTP2Stream]
40         self.events = {}  # type: typing.Dict[int, typing.List[h2.events.Event]]
41 
42         self.init_started = False
43 
44     @property
45     def is_http2(self) -> bool:
46         return True
47 
48     @property
49     def init_complete(self) -> BaseEvent:
50         # We do this lazily, to make sure backend autodetection always
51         # runs within an async context.
52         if not hasattr(self, "_initialization_complete"):
53             self._initialization_complete = self.backend.create_event()
54         return self._initialization_complete
55 
56     async def send(self, request: Request, timeout: Timeout = None) -> Response:
57         timeout = Timeout() if timeout is None else timeout
58 
59         if not self.init_started:
60             # The very first stream is responsible for initiating the connection.
61             self.init_started = True
62             await self.send_connection_init(timeout)
63             stream_id = self.state.get_next_available_stream_id()
64             self.init_complete.set()
65         else:
66             # All other streams need to wait until the connection is established.
67             await self.init_complete.wait()
68             stream_id = self.state.get_next_available_stream_id()
69 
70         stream = HTTP2Stream(stream_id=stream_id, connection=self)
71         self.streams[stream_id] = stream
72         self.events[stream_id] = []
73         return await stream.send(request, timeout)
74 
75     async def send_connection_init(self, timeout: Timeout) -> None:
76         """
77         The HTTP/2 connection requires some initial setup before we can start
78         using individual request/response streams on it.
79         """
80 
81         # Need to set these manually here instead of manipulating via
82         # __setitem__() otherwise the H2Connection will emit SettingsUpdate
83         # frames in addition to sending the undesired defaults.
84         self.state.local_settings = Settings(
85             client=True,
86             initial_values={
87                 # Disable PUSH_PROMISE frames from the server since we don't do anything
88                 # with them for now.  Maybe when we support caching?
89                 SettingCodes.ENABLE_PUSH: 0,
90                 # These two are taken from h2 for safe defaults
91                 SettingCodes.MAX_CONCURRENT_STREAMS: 100,
92                 SettingCodes.MAX_HEADER_LIST_SIZE: 65536,
93             },
94         )
95 
96         # Some websites (*cough* Yahoo *cough*) balk at this setting being
97         # present in the initial handshake since it's not defined in the original
98         # RFC despite the RFC mandating ignoring settings you don't know about.
99         del self.state.local_settings[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL]
100 
101         self.state.initiate_connection()
102         self.state.increment_flow_control_window(2 ** 24)
103         data_to_send = self.state.data_to_send()
104         await self.socket.write(data_to_send, timeout)
105 
106     @property
107     def is_closed(self) -> bool:
108         return False
109 
110     def is_connection_dropped(self) -> bool:
111         return self.socket.is_connection_dropped()
112 
113     async def close(self) -> None:
114         await self.socket.close()
115 
116     async def wait_for_outgoing_flow(self, stream_id: int, timeout: Timeout) -> int:
117         """
118         Returns the maximum allowable outgoing flow for a given stream.
119 
120         If the allowable flow is zero, then waits on the network until
121         WindowUpdated frames have increased the flow rate.
122 
123         https://tools.ietf.org/html/rfc7540#section-6.9
124         """
125         local_flow = self.state.local_flow_control_window(stream_id)
126         connection_flow = self.state.max_outbound_frame_size
127         flow = min(local_flow, connection_flow)
128         while flow == 0:
129             await self.receive_events(timeout)
130             local_flow = self.state.local_flow_control_window(stream_id)
131             connection_flow = self.state.max_outbound_frame_size
132             flow = min(local_flow, connection_flow)
133         return flow
134 
135     async def wait_for_event(self, stream_id: int, timeout: Timeout) -> h2.events.Event:
136         """
137         Returns the next event for a given stream.
138 
139         If no events are available yet, then waits on the network until
140         an event is available.
141         """
142         while not self.events[stream_id]:
143             await self.receive_events(timeout)
144         return self.events[stream_id].pop(0)
145 
146     async def receive_events(self, timeout: Timeout) -> None:
147         """
148         Read some data from the network, and update the H2 state.
149         """
150         data = await self.socket.read(self.READ_NUM_BYTES, timeout)
151         events = self.state.receive_data(data)
152         for event in events:
153             event_stream_id = getattr(event, "stream_id", 0)
154             logger.trace(f"receive_event stream_id={event_stream_id} event={event!r}")
155 
156             if hasattr(event, "error_code"):
157                 raise ProtocolError(event)
158 
159             if event_stream_id in self.events:
160                 self.events[event_stream_id].append(event)
161 
162         data_to_send = self.state.data_to_send()
163         await self.socket.write(data_to_send, timeout)
164 
165     async def send_headers(
166         self,
167         stream_id: int,
168         headers: typing.List[typing.Tuple[bytes, bytes]],
169         timeout: Timeout,
170     ) -> None:
171         self.state.send_headers(stream_id, headers)
172         self.state.increment_flow_control_window(2 ** 24, stream_id=stream_id)
173         data_to_send = self.state.data_to_send()
174         await self.socket.write(data_to_send, timeout)
175 
176     async def send_data(self, stream_id: int, chunk: bytes, timeout: Timeout) -> None:
177         self.state.send_data(stream_id, chunk)
178         data_to_send = self.state.data_to_send()
179         await self.socket.write(data_to_send, timeout)
180 
181     async def end_stream(self, stream_id: int, timeout: Timeout) -> None:
182         self.state.end_stream(stream_id)
183         data_to_send = self.state.data_to_send()
184         await self.socket.write(data_to_send, timeout)
185 
186     async def acknowledge_received_data(
187         self, stream_id: int, amount: int, timeout: Timeout
188     ) -> None:
189         self.state.acknowledge_received_data(amount, stream_id)
190         data_to_send = self.state.data_to_send()
191         await self.socket.write(data_to_send, timeout)
192 
193     async def close_stream(self, stream_id: int) -> None:
194         del self.streams[stream_id]
195         del self.events[stream_id]
196 
197         if not self.streams and self.on_release is not None:
198             await self.on_release()
199 
200 
201 class HTTP2Stream:
202     def __init__(self, stream_id: int, connection: HTTP2Connection) -> None:
203         self.stream_id = stream_id
204         self.connection = connection
205 
206     async def send(self, request: Request, timeout: Timeout) -> Response:
207         # Send the request.
208         await self.send_headers(request, timeout)
209         await self.send_body(request, timeout)
210 
211         # Receive the response.
212         status_code, headers = await self.receive_response(timeout)
213         stream = AsyncIteratorStream(
214             aiterator=self.body_iter(timeout), close_func=self.close
215         )
216 
217         return Response(
218             status_code=status_code,
219             http_version="HTTP/2",
220             headers=headers,
221             stream=stream,
222             request=request,
223         )
224 
225     async def send_headers(self, request: Request, timeout: Timeout) -> None:
226         headers = [
227             (b":method", request.method.encode("ascii")),
228             (b":authority", request.url.authority.encode("ascii")),
229             (b":scheme", request.url.scheme.encode("ascii")),
230             (b":path", request.url.full_path.encode("ascii")),
231         ] + [(k, v) for k, v in request.headers.raw if k != b"host"]
232 
233         logger.trace(
234             f"send_headers "
235             f"stream_id={self.stream_id} "
236             f"method={request.method!r} "
237             f"target={request.url.full_path!r} "
238             f"headers={headers!r}"
239         )
240         await self.connection.send_headers(self.stream_id, headers, timeout)
241 
242     async def send_body(self, request: Request, timeout: Timeout) -> None:
243         logger.trace(f"send_body stream_id={self.stream_id}")
244         async for data in request.stream:
245             while data:
246                 max_flow = await self.connection.wait_for_outgoing_flow(
247                     self.stream_id, timeout
248                 )
249                 chunk_size = min(len(data), max_flow)
250                 chunk, data = data[:chunk_size], data[chunk_size:]
251                 await self.connection.send_data(self.stream_id, chunk, timeout)
252 
253         await self.connection.end_stream(self.stream_id, timeout)
254 
255     async def receive_response(
256         self, timeout: Timeout
257     ) -> typing.Tuple[int, typing.List[typing.Tuple[bytes, bytes]]]:
258         """
259         Read the response status and headers from the network.
260         """
261         while True:
262             event = await self.connection.wait_for_event(self.stream_id, timeout)
263             if isinstance(event, h2.events.ResponseReceived):
264                 break
265 
266         status_code = 200
267         headers = []
268         for k, v in event.headers:
269             if k == b":status":
270                 status_code = int(v.decode("ascii", errors="ignore"))
271             elif not k.startswith(b":"):
272                 headers.append((k, v))
273 
274         return (status_code, headers)
275 
276     async def body_iter(self, timeout: Timeout) -> typing.AsyncIterator[bytes]:
277         while True:
278             event = await self.connection.wait_for_event(self.stream_id, timeout)
279             if isinstance(event, h2.events.DataReceived):
280                 amount = event.flow_controlled_length
281                 await self.connection.acknowledge_received_data(
282                     self.stream_id, amount, timeout
283                 )
284                 yield event.data
285             elif isinstance(event, (h2.events.StreamEnded, h2.events.StreamReset)):
286                 break
287 
288     async def close(self) -> None:
289         await self.connection.close_stream(self.stream_id)
290
```
--- 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/http2.py b/httpx/dispatch/http2.py
--- a/httpx/dispatch/http2.py
+++ b/httpx/dispatch/http2.py
@@ -166,9 +166,10 @@
         self,
         stream_id: int,
         headers: typing.List[typing.Tuple[bytes, bytes]],
+        end_stream: bool,
         timeout: Timeout,
     ) -> None:
-        self.state.send_headers(stream_id, headers)
+        self.state.send_headers(stream_id, headers, end_stream=end_stream)
         self.state.increment_flow_control_window(2 ** 24, stream_id=stream_id)
         data_to_send = self.state.data_to_send()
         await self.socket.write(data_to_send, timeout)
@@ -205,8 +206,14 @@
 
     async def send(self, request: Request, timeout: Timeout) -> Response:
         # Send the request.
-        await self.send_headers(request, timeout)
-        await self.send_body(request, timeout)
+        has_body = (
+            "Content-Length" in request.headers
+            or "Transfer-Encoding" in request.headers
+        )
+
+        await self.send_headers(request, has_body, timeout)
+        if has_body:
+            await self.send_body(request, timeout)
 
         # Receive the response.
         status_code, headers = await self.receive_response(timeout)
@@ -222,13 +229,20 @@
             request=request,
         )
 
-    async def send_headers(self, request: Request, timeout: Timeout) -> None:
+    async def send_headers(
+        self, request: Request, has_body: bool, timeout: Timeout
+    ) -> None:
         headers = [
             (b":method", request.method.encode("ascii")),
             (b":authority", request.url.authority.encode("ascii")),
             (b":scheme", request.url.scheme.encode("ascii")),
             (b":path", request.url.full_path.encode("ascii")),
-        ] + [(k, v) for k, v in request.headers.raw if k != b"host"]
+        ] + [
+            (k, v)
+            for k, v in request.headers.raw
+            if k not in (b"host", b"transfer-encoding")
+        ]
+        end_stream = not has_body
 
         logger.trace(
             f"send_headers "
@@ -237,7 +251,7 @@
             f"target={request.url.full_path!r} "
             f"headers={headers!r}"
         )
-        await self.connection.send_headers(self.stream_id, headers, timeout)
+        await self.connection.send_headers(self.stream_id, headers, end_stream, timeout)
 
     async def send_body(self, request: Request, timeout: Timeout) -> None:
         logger.trace(f"send_body stream_id={self.stream_id}")

{"golden_diff": "diff --git a/httpx/dispatch/http2.py b/httpx/dispatch/http2.py\n--- a/httpx/dispatch/http2.py\n+++ b/httpx/dispatch/http2.py\n@@ -166,9 +166,10 @@\n         self,\n         stream_id: int,\n         headers: typing.List[typing.Tuple[bytes, bytes]],\n+        end_stream: bool,\n         timeout: Timeout,\n     ) -> None:\n-        self.state.send_headers(stream_id, headers)\n+        self.state.send_headers(stream_id, headers, end_stream=end_stream)\n         self.state.increment_flow_control_window(2 ** 24, stream_id=stream_id)\n         data_to_send = self.state.data_to_send()\n         await self.socket.write(data_to_send, timeout)\n@@ -205,8 +206,14 @@\n \n     async def send(self, request: Request, timeout: Timeout) -> Response:\n         # Send the request.\n-        await self.send_headers(request, timeout)\n-        await self.send_body(request, timeout)\n+        has_body = (\n+            \"Content-Length\" in request.headers\n+            or \"Transfer-Encoding\" in request.headers\n+        )\n+\n+        await self.send_headers(request, has_body, timeout)\n+        if has_body:\n+            await self.send_body(request, timeout)\n \n         # Receive the response.\n         status_code, headers = await self.receive_response(timeout)\n@@ -222,13 +229,20 @@\n             request=request,\n         )\n \n-    async def send_headers(self, request: Request, timeout: Timeout) -> None:\n+    async def send_headers(\n+        self, request: Request, has_body: bool, timeout: Timeout\n+    ) -> None:\n         headers = [\n             (b\":method\", request.method.encode(\"ascii\")),\n             (b\":authority\", request.url.authority.encode(\"ascii\")),\n             (b\":scheme\", request.url.scheme.encode(\"ascii\")),\n             (b\":path\", request.url.full_path.encode(\"ascii\")),\n-        ] + [(k, v) for k, v in request.headers.raw if k != b\"host\"]\n+        ] + [\n+            (k, v)\n+            for k, v in request.headers.raw\n+            if k not in (b\"host\", b\"transfer-encoding\")\n+        ]\n+        end_stream = not has_body\n \n         logger.trace(\n             f\"send_headers \"\n@@ -237,7 +251,7 @@\n             f\"target={request.url.full_path!r} \"\n             f\"headers={headers!r}\"\n         )\n-        await self.connection.send_headers(self.stream_id, headers, timeout)\n+        await self.connection.send_headers(self.stream_id, headers, end_stream, timeout)\n \n     async def send_body(self, request: Request, timeout: Timeout) -> None:\n         logger.trace(f\"send_body stream_id={self.stream_id}\")\n", "issue": "Cannot connect to yahoo.com, receiving a ProtocolError on Stream 0 after connection\nCouple of issues here: we're not raising an exception after receiving a `ProtocolError` from HTTP/2 when connecting to `yahoo.com`. We should be bubbling that exception up to users.\r\n\r\nThe second issue is is we don't know why yahoo is giving us this error, we have to figure that out as I'm sure it'll impact more than just them.\r\n\r\nI found this by running interop tests against Alexa 1000 websites.\n", "before_files": [{"content": "import typing\n\nimport h2.connection\nimport h2.events\nfrom h2.config import H2Configuration\nfrom h2.settings import SettingCodes, Settings\n\nfrom ..backends.base import (\n    BaseEvent,\n    BaseSocketStream,\n    ConcurrencyBackend,\n    lookup_backend,\n)\nfrom ..config import Timeout\nfrom ..content_streams import AsyncIteratorStream\nfrom ..exceptions import ProtocolError\nfrom ..models import Request, Response\nfrom ..utils import get_logger\nfrom .base import OpenConnection\n\nlogger = get_logger(__name__)\n\n\nclass HTTP2Connection(OpenConnection):\n    READ_NUM_BYTES = 4096\n    CONFIG = H2Configuration(validate_inbound_headers=False)\n\n    def __init__(\n        self,\n        socket: BaseSocketStream,\n        backend: typing.Union[str, ConcurrencyBackend] = \"auto\",\n        on_release: typing.Callable = None,\n    ):\n        self.socket = socket\n        self.backend = lookup_backend(backend)\n        self.on_release = on_release\n        self.state = h2.connection.H2Connection(config=self.CONFIG)\n\n        self.streams = {}  # type: typing.Dict[int, HTTP2Stream]\n        self.events = {}  # type: typing.Dict[int, typing.List[h2.events.Event]]\n\n        self.init_started = False\n\n    @property\n    def is_http2(self) -> bool:\n        return True\n\n    @property\n    def init_complete(self) -> BaseEvent:\n        # We do this lazily, to make sure backend autodetection always\n        # runs within an async context.\n        if not hasattr(self, \"_initialization_complete\"):\n            self._initialization_complete = self.backend.create_event()\n        return self._initialization_complete\n\n    async def send(self, request: Request, timeout: Timeout = None) -> Response:\n        timeout = Timeout() if timeout is None else timeout\n\n        if not self.init_started:\n            # The very first stream is responsible for initiating the connection.\n            self.init_started = True\n            await self.send_connection_init(timeout)\n            stream_id = self.state.get_next_available_stream_id()\n            self.init_complete.set()\n        else:\n            # All other streams need to wait until the connection is established.\n            await self.init_complete.wait()\n            stream_id = self.state.get_next_available_stream_id()\n\n        stream = HTTP2Stream(stream_id=stream_id, connection=self)\n        self.streams[stream_id] = stream\n        self.events[stream_id] = []\n        return await stream.send(request, timeout)\n\n    async def send_connection_init(self, timeout: Timeout) -> None:\n        \"\"\"\n        The HTTP/2 connection requires some initial setup before we can start\n        using individual request/response streams on it.\n        \"\"\"\n\n        # Need to set these manually here instead of manipulating via\n        # __setitem__() otherwise the H2Connection will emit SettingsUpdate\n        # frames in addition to sending the undesired defaults.\n        self.state.local_settings = Settings(\n            client=True,\n            initial_values={\n                # Disable PUSH_PROMISE frames from the server since we don't do anything\n                # with them for now.  Maybe when we support caching?\n                SettingCodes.ENABLE_PUSH: 0,\n                # These two are taken from h2 for safe defaults\n                SettingCodes.MAX_CONCURRENT_STREAMS: 100,\n                SettingCodes.MAX_HEADER_LIST_SIZE: 65536,\n            },\n        )\n\n        # Some websites (*cough* Yahoo *cough*) balk at this setting being\n        # present in the initial handshake since it's not defined in the original\n        # RFC despite the RFC mandating ignoring settings you don't know about.\n        del self.state.local_settings[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL]\n\n        self.state.initiate_connection()\n        self.state.increment_flow_control_window(2 ** 24)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    @property\n    def is_closed(self) -> bool:\n        return False\n\n    def is_connection_dropped(self) -> bool:\n        return self.socket.is_connection_dropped()\n\n    async def close(self) -> None:\n        await self.socket.close()\n\n    async def wait_for_outgoing_flow(self, stream_id: int, timeout: Timeout) -> int:\n        \"\"\"\n        Returns the maximum allowable outgoing flow for a given stream.\n\n        If the allowable flow is zero, then waits on the network until\n        WindowUpdated frames have increased the flow rate.\n\n        https://tools.ietf.org/html/rfc7540#section-6.9\n        \"\"\"\n        local_flow = self.state.local_flow_control_window(stream_id)\n        connection_flow = self.state.max_outbound_frame_size\n        flow = min(local_flow, connection_flow)\n        while flow == 0:\n            await self.receive_events(timeout)\n            local_flow = self.state.local_flow_control_window(stream_id)\n            connection_flow = self.state.max_outbound_frame_size\n            flow = min(local_flow, connection_flow)\n        return flow\n\n    async def wait_for_event(self, stream_id: int, timeout: Timeout) -> h2.events.Event:\n        \"\"\"\n        Returns the next event for a given stream.\n\n        If no events are available yet, then waits on the network until\n        an event is available.\n        \"\"\"\n        while not self.events[stream_id]:\n            await self.receive_events(timeout)\n        return self.events[stream_id].pop(0)\n\n    async def receive_events(self, timeout: Timeout) -> None:\n        \"\"\"\n        Read some data from the network, and update the H2 state.\n        \"\"\"\n        data = await self.socket.read(self.READ_NUM_BYTES, timeout)\n        events = self.state.receive_data(data)\n        for event in events:\n            event_stream_id = getattr(event, \"stream_id\", 0)\n            logger.trace(f\"receive_event stream_id={event_stream_id} event={event!r}\")\n\n            if hasattr(event, \"error_code\"):\n                raise ProtocolError(event)\n\n            if event_stream_id in self.events:\n                self.events[event_stream_id].append(event)\n\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def send_headers(\n        self,\n        stream_id: int,\n        headers: typing.List[typing.Tuple[bytes, bytes]],\n        timeout: Timeout,\n    ) -> None:\n        self.state.send_headers(stream_id, headers)\n        self.state.increment_flow_control_window(2 ** 24, stream_id=stream_id)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def send_data(self, stream_id: int, chunk: bytes, timeout: Timeout) -> None:\n        self.state.send_data(stream_id, chunk)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def end_stream(self, stream_id: int, timeout: Timeout) -> None:\n        self.state.end_stream(stream_id)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def acknowledge_received_data(\n        self, stream_id: int, amount: int, timeout: Timeout\n    ) -> None:\n        self.state.acknowledge_received_data(amount, stream_id)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def close_stream(self, stream_id: int) -> None:\n        del self.streams[stream_id]\n        del self.events[stream_id]\n\n        if not self.streams and self.on_release is not None:\n            await self.on_release()\n\n\nclass HTTP2Stream:\n    def __init__(self, stream_id: int, connection: HTTP2Connection) -> None:\n        self.stream_id = stream_id\n        self.connection = connection\n\n    async def send(self, request: Request, timeout: Timeout) -> Response:\n        # Send the request.\n        await self.send_headers(request, timeout)\n        await self.send_body(request, timeout)\n\n        # Receive the response.\n        status_code, headers = await self.receive_response(timeout)\n        stream = AsyncIteratorStream(\n            aiterator=self.body_iter(timeout), close_func=self.close\n        )\n\n        return Response(\n            status_code=status_code,\n            http_version=\"HTTP/2\",\n            headers=headers,\n            stream=stream,\n            request=request,\n        )\n\n    async def send_headers(self, request: Request, timeout: Timeout) -> None:\n        headers = [\n            (b\":method\", request.method.encode(\"ascii\")),\n            (b\":authority\", request.url.authority.encode(\"ascii\")),\n            (b\":scheme\", request.url.scheme.encode(\"ascii\")),\n            (b\":path\", request.url.full_path.encode(\"ascii\")),\n        ] + [(k, v) for k, v in request.headers.raw if k != b\"host\"]\n\n        logger.trace(\n            f\"send_headers \"\n            f\"stream_id={self.stream_id} \"\n            f\"method={request.method!r} \"\n            f\"target={request.url.full_path!r} \"\n            f\"headers={headers!r}\"\n        )\n        await self.connection.send_headers(self.stream_id, headers, timeout)\n\n    async def send_body(self, request: Request, timeout: Timeout) -> None:\n        logger.trace(f\"send_body stream_id={self.stream_id}\")\n        async for data in request.stream:\n            while data:\n                max_flow = await self.connection.wait_for_outgoing_flow(\n                    self.stream_id, timeout\n                )\n                chunk_size = min(len(data), max_flow)\n                chunk, data = data[:chunk_size], data[chunk_size:]\n                await self.connection.send_data(self.stream_id, chunk, timeout)\n\n        await self.connection.end_stream(self.stream_id, timeout)\n\n    async def receive_response(\n        self, timeout: Timeout\n    ) -> typing.Tuple[int, typing.List[typing.Tuple[bytes, bytes]]]:\n        \"\"\"\n        Read the response status and headers from the network.\n        \"\"\"\n        while True:\n            event = await self.connection.wait_for_event(self.stream_id, timeout)\n            if isinstance(event, h2.events.ResponseReceived):\n                break\n\n        status_code = 200\n        headers = []\n        for k, v in event.headers:\n            if k == b\":status\":\n                status_code = int(v.decode(\"ascii\", errors=\"ignore\"))\n            elif not k.startswith(b\":\"):\n                headers.append((k, v))\n\n        return (status_code, headers)\n\n    async def body_iter(self, timeout: Timeout) -> typing.AsyncIterator[bytes]:\n        while True:\n            event = await self.connection.wait_for_event(self.stream_id, timeout)\n            if isinstance(event, h2.events.DataReceived):\n                amount = event.flow_controlled_length\n                await self.connection.acknowledge_received_data(\n                    self.stream_id, amount, timeout\n                )\n                yield event.data\n            elif isinstance(event, (h2.events.StreamEnded, h2.events.StreamReset)):\n                break\n\n    async def close(self) -> None:\n        await self.connection.close_stream(self.stream_id)\n", "path": "httpx/dispatch/http2.py"}], "after_files": [{"content": "import typing\n\nimport h2.connection\nimport h2.events\nfrom h2.config import H2Configuration\nfrom h2.settings import SettingCodes, Settings\n\nfrom ..backends.base import (\n    BaseEvent,\n    BaseSocketStream,\n    ConcurrencyBackend,\n    lookup_backend,\n)\nfrom ..config import Timeout\nfrom ..content_streams import AsyncIteratorStream\nfrom ..exceptions import ProtocolError\nfrom ..models import Request, Response\nfrom ..utils import get_logger\nfrom .base import OpenConnection\n\nlogger = get_logger(__name__)\n\n\nclass HTTP2Connection(OpenConnection):\n    READ_NUM_BYTES = 4096\n    CONFIG = H2Configuration(validate_inbound_headers=False)\n\n    def __init__(\n        self,\n        socket: BaseSocketStream,\n        backend: typing.Union[str, ConcurrencyBackend] = \"auto\",\n        on_release: typing.Callable = None,\n    ):\n        self.socket = socket\n        self.backend = lookup_backend(backend)\n        self.on_release = on_release\n        self.state = h2.connection.H2Connection(config=self.CONFIG)\n\n        self.streams = {}  # type: typing.Dict[int, HTTP2Stream]\n        self.events = {}  # type: typing.Dict[int, typing.List[h2.events.Event]]\n\n        self.init_started = False\n\n    @property\n    def is_http2(self) -> bool:\n        return True\n\n    @property\n    def init_complete(self) -> BaseEvent:\n        # We do this lazily, to make sure backend autodetection always\n        # runs within an async context.\n        if not hasattr(self, \"_initialization_complete\"):\n            self._initialization_complete = self.backend.create_event()\n        return self._initialization_complete\n\n    async def send(self, request: Request, timeout: Timeout = None) -> Response:\n        timeout = Timeout() if timeout is None else timeout\n\n        if not self.init_started:\n            # The very first stream is responsible for initiating the connection.\n            self.init_started = True\n            await self.send_connection_init(timeout)\n            stream_id = self.state.get_next_available_stream_id()\n            self.init_complete.set()\n        else:\n            # All other streams need to wait until the connection is established.\n            await self.init_complete.wait()\n            stream_id = self.state.get_next_available_stream_id()\n\n        stream = HTTP2Stream(stream_id=stream_id, connection=self)\n        self.streams[stream_id] = stream\n        self.events[stream_id] = []\n        return await stream.send(request, timeout)\n\n    async def send_connection_init(self, timeout: Timeout) -> None:\n        \"\"\"\n        The HTTP/2 connection requires some initial setup before we can start\n        using individual request/response streams on it.\n        \"\"\"\n\n        # Need to set these manually here instead of manipulating via\n        # __setitem__() otherwise the H2Connection will emit SettingsUpdate\n        # frames in addition to sending the undesired defaults.\n        self.state.local_settings = Settings(\n            client=True,\n            initial_values={\n                # Disable PUSH_PROMISE frames from the server since we don't do anything\n                # with them for now.  Maybe when we support caching?\n                SettingCodes.ENABLE_PUSH: 0,\n                # These two are taken from h2 for safe defaults\n                SettingCodes.MAX_CONCURRENT_STREAMS: 100,\n                SettingCodes.MAX_HEADER_LIST_SIZE: 65536,\n            },\n        )\n\n        # Some websites (*cough* Yahoo *cough*) balk at this setting being\n        # present in the initial handshake since it's not defined in the original\n        # RFC despite the RFC mandating ignoring settings you don't know about.\n        del self.state.local_settings[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL]\n\n        self.state.initiate_connection()\n        self.state.increment_flow_control_window(2 ** 24)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    @property\n    def is_closed(self) -> bool:\n        return False\n\n    def is_connection_dropped(self) -> bool:\n        return self.socket.is_connection_dropped()\n\n    async def close(self) -> None:\n        await self.socket.close()\n\n    async def wait_for_outgoing_flow(self, stream_id: int, timeout: Timeout) -> int:\n        \"\"\"\n        Returns the maximum allowable outgoing flow for a given stream.\n\n        If the allowable flow is zero, then waits on the network until\n        WindowUpdated frames have increased the flow rate.\n\n        https://tools.ietf.org/html/rfc7540#section-6.9\n        \"\"\"\n        local_flow = self.state.local_flow_control_window(stream_id)\n        connection_flow = self.state.max_outbound_frame_size\n        flow = min(local_flow, connection_flow)\n        while flow == 0:\n            await self.receive_events(timeout)\n            local_flow = self.state.local_flow_control_window(stream_id)\n            connection_flow = self.state.max_outbound_frame_size\n            flow = min(local_flow, connection_flow)\n        return flow\n\n    async def wait_for_event(self, stream_id: int, timeout: Timeout) -> h2.events.Event:\n        \"\"\"\n        Returns the next event for a given stream.\n\n        If no events are available yet, then waits on the network until\n        an event is available.\n        \"\"\"\n        while not self.events[stream_id]:\n            await self.receive_events(timeout)\n        return self.events[stream_id].pop(0)\n\n    async def receive_events(self, timeout: Timeout) -> None:\n        \"\"\"\n        Read some data from the network, and update the H2 state.\n        \"\"\"\n        data = await self.socket.read(self.READ_NUM_BYTES, timeout)\n        events = self.state.receive_data(data)\n        for event in events:\n            event_stream_id = getattr(event, \"stream_id\", 0)\n            logger.trace(f\"receive_event stream_id={event_stream_id} event={event!r}\")\n\n            if hasattr(event, \"error_code\"):\n                raise ProtocolError(event)\n\n            if event_stream_id in self.events:\n                self.events[event_stream_id].append(event)\n\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def send_headers(\n        self,\n        stream_id: int,\n        headers: typing.List[typing.Tuple[bytes, bytes]],\n        end_stream: bool,\n        timeout: Timeout,\n    ) -> None:\n        self.state.send_headers(stream_id, headers, end_stream=end_stream)\n        self.state.increment_flow_control_window(2 ** 24, stream_id=stream_id)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def send_data(self, stream_id: int, chunk: bytes, timeout: Timeout) -> None:\n        self.state.send_data(stream_id, chunk)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def end_stream(self, stream_id: int, timeout: Timeout) -> None:\n        self.state.end_stream(stream_id)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def acknowledge_received_data(\n        self, stream_id: int, amount: int, timeout: Timeout\n    ) -> None:\n        self.state.acknowledge_received_data(amount, stream_id)\n        data_to_send = self.state.data_to_send()\n        await self.socket.write(data_to_send, timeout)\n\n    async def close_stream(self, stream_id: int) -> None:\n        del self.streams[stream_id]\n        del self.events[stream_id]\n\n        if not self.streams and self.on_release is not None:\n            await self.on_release()\n\n\nclass HTTP2Stream:\n    def __init__(self, stream_id: int, connection: HTTP2Connection) -> None:\n        self.stream_id = stream_id\n        self.connection = connection\n\n    async def send(self, request: Request, timeout: Timeout) -> Response:\n        # Send the request.\n        has_body = (\n            \"Content-Length\" in request.headers\n            or \"Transfer-Encoding\" in request.headers\n        )\n\n        await self.send_headers(request, has_body, timeout)\n        if has_body:\n            await self.send_body(request, timeout)\n\n        # Receive the response.\n        status_code, headers = await self.receive_response(timeout)\n        stream = AsyncIteratorStream(\n            aiterator=self.body_iter(timeout), close_func=self.close\n        )\n\n        return Response(\n            status_code=status_code,\n            http_version=\"HTTP/2\",\n            headers=headers,\n            stream=stream,\n            request=request,\n        )\n\n    async def send_headers(\n        self, request: Request, has_body: bool, timeout: Timeout\n    ) -> None:\n        headers = [\n            (b\":method\", request.method.encode(\"ascii\")),\n            (b\":authority\", request.url.authority.encode(\"ascii\")),\n            (b\":scheme\", request.url.scheme.encode(\"ascii\")),\n            (b\":path\", request.url.full_path.encode(\"ascii\")),\n        ] + [\n            (k, v)\n            for k, v in request.headers.raw\n            if k not in (b\"host\", b\"transfer-encoding\")\n        ]\n        end_stream = not has_body\n\n        logger.trace(\n            f\"send_headers \"\n            f\"stream_id={self.stream_id} \"\n            f\"method={request.method!r} \"\n            f\"target={request.url.full_path!r} \"\n            f\"headers={headers!r}\"\n        )\n        await self.connection.send_headers(self.stream_id, headers, end_stream, timeout)\n\n    async def send_body(self, request: Request, timeout: Timeout) -> None:\n        logger.trace(f\"send_body stream_id={self.stream_id}\")\n        async for data in request.stream:\n            while data:\n                max_flow = await self.connection.wait_for_outgoing_flow(\n                    self.stream_id, timeout\n                )\n                chunk_size = min(len(data), max_flow)\n                chunk, data = data[:chunk_size], data[chunk_size:]\n                await self.connection.send_data(self.stream_id, chunk, timeout)\n\n        await self.connection.end_stream(self.stream_id, timeout)\n\n    async def receive_response(\n        self, timeout: Timeout\n    ) -> typing.Tuple[int, typing.List[typing.Tuple[bytes, bytes]]]:\n        \"\"\"\n        Read the response status and headers from the network.\n        \"\"\"\n        while True:\n            event = await self.connection.wait_for_event(self.stream_id, timeout)\n            if isinstance(event, h2.events.ResponseReceived):\n                break\n\n        status_code = 200\n        headers = []\n        for k, v in event.headers:\n            if k == b\":status\":\n                status_code = int(v.decode(\"ascii\", errors=\"ignore\"))\n            elif not k.startswith(b\":\"):\n                headers.append((k, v))\n\n        return (status_code, headers)\n\n    async def body_iter(self, timeout: Timeout) -> typing.AsyncIterator[bytes]:\n        while True:\n            event = await self.connection.wait_for_event(self.stream_id, timeout)\n            if isinstance(event, h2.events.DataReceived):\n                amount = event.flow_controlled_length\n                await self.connection.acknowledge_received_data(\n                    self.stream_id, amount, timeout\n                )\n                yield event.data\n            elif isinstance(event, (h2.events.StreamEnded, h2.events.StreamReset)):\n                break\n\n    async def close(self) -> None:\n        await self.connection.close_stream(self.stream_id)\n", "path": "httpx/dispatch/http2.py"}]}