We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
header value surrounded by whitespace
httpx [doesn't accept header value surrounded by whitespace](https://github.com/python-hyper/hyper-h2/blob/13005074d14c7d32f8eaf1683b854446a09d09d3/h2/utilities.py#L265), but other http2 implementations seems to accept them (nghttp2, browsers Firefox, Chrome).

Example:
```
async with httpx.Client(http2=True) as client:
	response = await client.get('https://a.searx.space/headervaluetest')
```

Result :
```
ProtocolError: Received header value surrounded by whitespace b'A value '
```

The nghttp2  implementation seems here:
* https://github.com/nghttp2/nghttp2/blob/bb519154fe62f7ff7e5eb7269e05043dec6d3682/lib/nghttp2_http.c#L332
* https://github.com/nghttp2/nghttp2/blob/bb519154fe62f7ff7e5eb7269e05043dec6d3682/lib/nghttp2_helper.c#L498

I'm not sure if it should be declared as a bug or not.

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

{"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@@ -2,6 +2,7 @@\n \n import h2.connection\n import h2.events\n+from h2.config import H2Configuration\n from h2.settings import SettingCodes, Settings\n \n from ..concurrency.base import (\n@@ -21,6 +22,7 @@\n \n class HTTP2Connection(OpenConnection):\n     READ_NUM_BYTES = 4096\n+    CONFIG = H2Configuration(validate_inbound_headers=False)\n \n     def __init__(\n         self,\n@@ -31,7 +33,7 @@\n         self.socket = socket\n         self.backend = lookup_backend(backend)\n         self.on_release = on_release\n-        self.state = h2.connection.H2Connection()\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", "issue": "header value surrounded by whitespace\nhttpx [doesn't accept header value surrounded by whitespace](https://github.com/python-hyper/hyper-h2/blob/13005074d14c7d32f8eaf1683b854446a09d09d3/h2/utilities.py#L265), but other http2 implementations seems to accept them (nghttp2, browsers Firefox, Chrome).\r\n\r\nExample:\r\n```\r\nasync with httpx.Client(http2=True) as client:\r\n\tresponse = await client.get('https://a.searx.space/headervaluetest')\r\n```\r\n\r\nResult :\r\n```\r\nProtocolError: Received header value surrounded by whitespace b'A value '\r\n```\r\n\r\nThe nghttp2  implementation seems here:\r\n* https://github.com/nghttp2/nghttp2/blob/bb519154fe62f7ff7e5eb7269e05043dec6d3682/lib/nghttp2_http.c#L332\r\n* https://github.com/nghttp2/nghttp2/blob/bb519154fe62f7ff7e5eb7269e05043dec6d3682/lib/nghttp2_helper.c#L498\r\n\r\nI'm not sure if it should be declared as a bug or not.\n", "before_files": [{"content": "import typing\n\nimport h2.connection\nimport h2.events\nfrom h2.settings import SettingCodes, Settings\n\nfrom ..concurrency.base import (\n    BaseEvent,\n    BaseSocketStream,\n    ConcurrencyBackend,\n    lookup_backend,\n)\nfrom ..config import Timeout\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\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()\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        content = self.body_iter(timeout)\n        return Response(\n            status_code=status_code,\n            http_version=\"HTTP/2\",\n            headers=headers,\n            content=content,\n            on_close=self.close,\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 ..concurrency.base import (\n    BaseEvent,\n    BaseSocketStream,\n    ConcurrencyBackend,\n    lookup_backend,\n)\nfrom ..config import Timeout\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        content = self.body_iter(timeout)\n        return Response(\n            status_code=status_code,\n            http_version=\"HTTP/2\",\n            headers=headers,\n            content=content,\n            on_close=self.close,\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"}]}