We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
LineDecoder is accidentally quadratic: iter_lines() seems to hang forever
When calling `Response.iter_lines()`, things can seem to hang forever.

The problem is that `LineDecoder` is quadratic in it's string copying behaviour. If a 31MB chunk with 18,768 lines is passed in to `LineDecoder()` then it takes 1m45s to process it vs 0.1s for a simple `text.splitlines()`. This is readily reproduced by simply using `LineDecoder()` to decode a file, like `/usr/share/dict/words` (do not attempt, you may be waiting until the heat death of the universe).

It may be a bug somewhere else that `iter_text()` is returning chunks that are too big.

But either way, you probably want to reconsider the wisdom of slicing the string to chop the beginning off it vs. just keeping an index.

You can slice the string before returning from the decoder so it's only done once per chunk.

Anyhoo, thanks for the great work!

--- 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/_decoders.py`
Content:
```
1 """
2 Handlers for Content-Encoding.
3 
4 See: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Encoding
5 """
6 import codecs
7 import io
8 import typing
9 import zlib
10 
11 from ._compat import brotli
12 from ._exceptions import DecodingError
13 
14 
15 class ContentDecoder:
16     def decode(self, data: bytes) -> bytes:
17         raise NotImplementedError()  # pragma: no cover
18 
19     def flush(self) -> bytes:
20         raise NotImplementedError()  # pragma: no cover
21 
22 
23 class IdentityDecoder(ContentDecoder):
24     """
25     Handle unencoded data.
26     """
27 
28     def decode(self, data: bytes) -> bytes:
29         return data
30 
31     def flush(self) -> bytes:
32         return b""
33 
34 
35 class DeflateDecoder(ContentDecoder):
36     """
37     Handle 'deflate' decoding.
38 
39     See: https://stackoverflow.com/questions/1838699
40     """
41 
42     def __init__(self) -> None:
43         self.first_attempt = True
44         self.decompressor = zlib.decompressobj()
45 
46     def decode(self, data: bytes) -> bytes:
47         was_first_attempt = self.first_attempt
48         self.first_attempt = False
49         try:
50             return self.decompressor.decompress(data)
51         except zlib.error as exc:
52             if was_first_attempt:
53                 self.decompressor = zlib.decompressobj(-zlib.MAX_WBITS)
54                 return self.decode(data)
55             raise DecodingError(str(exc)) from exc
56 
57     def flush(self) -> bytes:
58         try:
59             return self.decompressor.flush()
60         except zlib.error as exc:  # pragma: no cover
61             raise DecodingError(str(exc)) from exc
62 
63 
64 class GZipDecoder(ContentDecoder):
65     """
66     Handle 'gzip' decoding.
67 
68     See: https://stackoverflow.com/questions/1838699
69     """
70 
71     def __init__(self) -> None:
72         self.decompressor = zlib.decompressobj(zlib.MAX_WBITS | 16)
73 
74     def decode(self, data: bytes) -> bytes:
75         try:
76             return self.decompressor.decompress(data)
77         except zlib.error as exc:
78             raise DecodingError(str(exc)) from exc
79 
80     def flush(self) -> bytes:
81         try:
82             return self.decompressor.flush()
83         except zlib.error as exc:  # pragma: no cover
84             raise DecodingError(str(exc)) from exc
85 
86 
87 class BrotliDecoder(ContentDecoder):
88     """
89     Handle 'brotli' decoding.
90 
91     Requires `pip install brotlipy`. See: https://brotlipy.readthedocs.io/
92         or   `pip install brotli`. See https://github.com/google/brotli
93     Supports both 'brotlipy' and 'Brotli' packages since they share an import
94     name. The top branches are for 'brotlipy' and bottom branches for 'Brotli'
95     """
96 
97     def __init__(self) -> None:
98         if brotli is None:  # pragma: no cover
99             raise ImportError(
100                 "Using 'BrotliDecoder', but neither of the 'brotlicffi' or 'brotli' "
101                 "packages have been installed. "
102                 "Make sure to install httpx using `pip install httpx[brotli]`."
103             ) from None
104 
105         self.decompressor = brotli.Decompressor()
106         self.seen_data = False
107         self._decompress: typing.Callable[[bytes], bytes]
108         if hasattr(self.decompressor, "decompress"):
109             # The 'brotlicffi' package.
110             self._decompress = self.decompressor.decompress  # pragma: no cover
111         else:
112             # The 'brotli' package.
113             self._decompress = self.decompressor.process  # pragma: no cover
114 
115     def decode(self, data: bytes) -> bytes:
116         if not data:
117             return b""
118         self.seen_data = True
119         try:
120             return self._decompress(data)
121         except brotli.error as exc:
122             raise DecodingError(str(exc)) from exc
123 
124     def flush(self) -> bytes:
125         if not self.seen_data:
126             return b""
127         try:
128             if hasattr(self.decompressor, "finish"):
129                 # Only available in the 'brotlicffi' package.
130 
131                 # As the decompressor decompresses eagerly, this
132                 # will never actually emit any data. However, it will potentially throw
133                 # errors if a truncated or damaged data stream has been used.
134                 self.decompressor.finish()  # pragma: no cover
135             return b""
136         except brotli.error as exc:  # pragma: no cover
137             raise DecodingError(str(exc)) from exc
138 
139 
140 class MultiDecoder(ContentDecoder):
141     """
142     Handle the case where multiple encodings have been applied.
143     """
144 
145     def __init__(self, children: typing.Sequence[ContentDecoder]) -> None:
146         """
147         'children' should be a sequence of decoders in the order in which
148         each was applied.
149         """
150         # Note that we reverse the order for decoding.
151         self.children = list(reversed(children))
152 
153     def decode(self, data: bytes) -> bytes:
154         for child in self.children:
155             data = child.decode(data)
156         return data
157 
158     def flush(self) -> bytes:
159         data = b""
160         for child in self.children:
161             data = child.decode(data) + child.flush()
162         return data
163 
164 
165 class ByteChunker:
166     """
167     Handles returning byte content in fixed-size chunks.
168     """
169 
170     def __init__(self, chunk_size: typing.Optional[int] = None) -> None:
171         self._buffer = io.BytesIO()
172         self._chunk_size = chunk_size
173 
174     def decode(self, content: bytes) -> typing.List[bytes]:
175         if self._chunk_size is None:
176             return [content] if content else []
177 
178         self._buffer.write(content)
179         if self._buffer.tell() >= self._chunk_size:
180             value = self._buffer.getvalue()
181             chunks = [
182                 value[i : i + self._chunk_size]
183                 for i in range(0, len(value), self._chunk_size)
184             ]
185             if len(chunks[-1]) == self._chunk_size:
186                 self._buffer.seek(0)
187                 self._buffer.truncate()
188                 return chunks
189             else:
190                 self._buffer.seek(0)
191                 self._buffer.write(chunks[-1])
192                 self._buffer.truncate()
193                 return chunks[:-1]
194         else:
195             return []
196 
197     def flush(self) -> typing.List[bytes]:
198         value = self._buffer.getvalue()
199         self._buffer.seek(0)
200         self._buffer.truncate()
201         return [value] if value else []
202 
203 
204 class TextChunker:
205     """
206     Handles returning text content in fixed-size chunks.
207     """
208 
209     def __init__(self, chunk_size: typing.Optional[int] = None) -> None:
210         self._buffer = io.StringIO()
211         self._chunk_size = chunk_size
212 
213     def decode(self, content: str) -> typing.List[str]:
214         if self._chunk_size is None:
215             return [content]
216 
217         self._buffer.write(content)
218         if self._buffer.tell() >= self._chunk_size:
219             value = self._buffer.getvalue()
220             chunks = [
221                 value[i : i + self._chunk_size]
222                 for i in range(0, len(value), self._chunk_size)
223             ]
224             if len(chunks[-1]) == self._chunk_size:
225                 self._buffer.seek(0)
226                 self._buffer.truncate()
227                 return chunks
228             else:
229                 self._buffer.seek(0)
230                 self._buffer.write(chunks[-1])
231                 self._buffer.truncate()
232                 return chunks[:-1]
233         else:
234             return []
235 
236     def flush(self) -> typing.List[str]:
237         value = self._buffer.getvalue()
238         self._buffer.seek(0)
239         self._buffer.truncate()
240         return [value] if value else []
241 
242 
243 class TextDecoder:
244     """
245     Handles incrementally decoding bytes into text
246     """
247 
248     def __init__(self, encoding: str = "utf-8"):
249         self.decoder = codecs.getincrementaldecoder(encoding)(errors="replace")
250 
251     def decode(self, data: bytes) -> str:
252         return self.decoder.decode(data)
253 
254     def flush(self) -> str:
255         return self.decoder.decode(b"", True)
256 
257 
258 class LineDecoder:
259     """
260     Handles incrementally reading lines from text.
261 
262     Uses universal line decoding, supporting any of `\n`, `\r`, or `\r\n`
263     as line endings, normalizing to `\n`.
264     """
265 
266     def __init__(self) -> None:
267         self.buffer = ""
268 
269     def decode(self, text: str) -> typing.List[str]:
270         lines = []
271 
272         if text and self.buffer and self.buffer[-1] == "\r":
273             if text.startswith("\n"):
274                 # Handle the case where we have an "\r\n" split across
275                 # our previous input, and our new chunk.
276                 lines.append(self.buffer[:-1] + "\n")
277                 self.buffer = ""
278                 text = text[1:]
279             else:
280                 # Handle the case where we have "\r" at the end of our
281                 # previous input.
282                 lines.append(self.buffer[:-1] + "\n")
283                 self.buffer = ""
284 
285         while text:
286             num_chars = len(text)
287             for idx in range(num_chars):
288                 char = text[idx]
289                 next_char = None if idx + 1 == num_chars else text[idx + 1]
290                 if char == "\n":
291                     lines.append(self.buffer + text[: idx + 1])
292                     self.buffer = ""
293                     text = text[idx + 1 :]
294                     break
295                 elif char == "\r" and next_char == "\n":
296                     lines.append(self.buffer + text[:idx] + "\n")
297                     self.buffer = ""
298                     text = text[idx + 2 :]
299                     break
300                 elif char == "\r" and next_char is not None:
301                     lines.append(self.buffer + text[:idx] + "\n")
302                     self.buffer = ""
303                     text = text[idx + 1 :]
304                     break
305                 elif next_char is None:
306                     self.buffer += text
307                     text = ""
308                     break
309 
310         return lines
311 
312     def flush(self) -> typing.List[str]:
313         if self.buffer.endswith("\r"):
314             # Handle the case where we had a trailing '\r', which could have
315             # been a '\r\n' pair.
316             lines = [self.buffer[:-1] + "\n"]
317         elif self.buffer:
318             lines = [self.buffer]
319         else:
320             lines = []
321         self.buffer = ""
322         return lines
323 
324 
325 SUPPORTED_DECODERS = {
326     "identity": IdentityDecoder,
327     "gzip": GZipDecoder,
328     "deflate": DeflateDecoder,
329     "br": BrotliDecoder,
330 }
331 
332 
333 if brotli is None:
334     SUPPORTED_DECODERS.pop("br")  # pragma: no cover
335
```
--- 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/_decoders.py b/httpx/_decoders.py
--- a/httpx/_decoders.py
+++ b/httpx/_decoders.py
@@ -259,66 +259,56 @@
     """
     Handles incrementally reading lines from text.
 
-    Uses universal line decoding, supporting any of `\n`, `\r`, or `\r\n`
-    as line endings, normalizing to `\n`.
+    Has the same behaviour as the stdllib splitlines, but handling the input iteratively.
     """
 
     def __init__(self) -> None:
-        self.buffer = ""
+        self.buffer: typing.List[str] = []
+        self.trailing_cr: bool = False
 
     def decode(self, text: str) -> typing.List[str]:
-        lines = []
-
-        if text and self.buffer and self.buffer[-1] == "\r":
-            if text.startswith("\n"):
-                # Handle the case where we have an "\r\n" split across
-                # our previous input, and our new chunk.
-                lines.append(self.buffer[:-1] + "\n")
-                self.buffer = ""
-                text = text[1:]
-            else:
-                # Handle the case where we have "\r" at the end of our
-                # previous input.
-                lines.append(self.buffer[:-1] + "\n")
-                self.buffer = ""
-
-        while text:
-            num_chars = len(text)
-            for idx in range(num_chars):
-                char = text[idx]
-                next_char = None if idx + 1 == num_chars else text[idx + 1]
-                if char == "\n":
-                    lines.append(self.buffer + text[: idx + 1])
-                    self.buffer = ""
-                    text = text[idx + 1 :]
-                    break
-                elif char == "\r" and next_char == "\n":
-                    lines.append(self.buffer + text[:idx] + "\n")
-                    self.buffer = ""
-                    text = text[idx + 2 :]
-                    break
-                elif char == "\r" and next_char is not None:
-                    lines.append(self.buffer + text[:idx] + "\n")
-                    self.buffer = ""
-                    text = text[idx + 1 :]
-                    break
-                elif next_char is None:
-                    self.buffer += text
-                    text = ""
-                    break
+        # See https://docs.python.org/3/library/stdtypes.html#str.splitlines
+        NEWLINE_CHARS = "\n\r\x0b\x0c\x1c\x1d\x1e\x85\u2028\u2029"
+
+        # We always push a trailing `\r` into the next decode iteration.
+        if self.trailing_cr:
+            text = "\r" + text
+            self.trailing_cr = False
+        if text.endswith("\r"):
+            self.trailing_cr = True
+            text = text[:-1]
+
+        if not text:
+            return []
+
+        trailing_newline = text[-1] in NEWLINE_CHARS
+        lines = text.splitlines()
+
+        if len(lines) == 1 and not trailing_newline:
+            # No new lines, buffer the input and continue.
+            self.buffer.append(lines[0])
+            return []
+
+        if self.buffer:
+            # Include any existing buffer in the first portion of the
+            # splitlines result.
+            lines = ["".join(self.buffer) + lines[0]] + lines[1:]
+            self.buffer = []
+
+        if not trailing_newline:
+            # If the last segment of splitlines is not newline terminated,
+            # then drop it from our output and start a new buffer.
+            self.buffer = [lines.pop()]
 
         return lines
 
     def flush(self) -> typing.List[str]:
-        if self.buffer.endswith("\r"):
-            # Handle the case where we had a trailing '\r', which could have
-            # been a '\r\n' pair.
-            lines = [self.buffer[:-1] + "\n"]
-        elif self.buffer:
-            lines = [self.buffer]
-        else:
-            lines = []
-        self.buffer = ""
+        if not self.buffer and not self.trailing_cr:
+            return []
+
+        lines = ["".join(self.buffer)]
+        self.buffer = []
+        self.trailing_cr = False
         return lines

{"golden_diff": "diff --git a/httpx/_decoders.py b/httpx/_decoders.py\n--- a/httpx/_decoders.py\n+++ b/httpx/_decoders.py\n@@ -259,66 +259,56 @@\n     \"\"\"\n     Handles incrementally reading lines from text.\n \n-    Uses universal line decoding, supporting any of `\\n`, `\\r`, or `\\r\\n`\n-    as line endings, normalizing to `\\n`.\n+    Has the same behaviour as the stdllib splitlines, but handling the input iteratively.\n     \"\"\"\n \n     def __init__(self) -> None:\n-        self.buffer = \"\"\n+        self.buffer: typing.List[str] = []\n+        self.trailing_cr: bool = False\n \n     def decode(self, text: str) -> typing.List[str]:\n-        lines = []\n-\n-        if text and self.buffer and self.buffer[-1] == \"\\r\":\n-            if text.startswith(\"\\n\"):\n-                # Handle the case where we have an \"\\r\\n\" split across\n-                # our previous input, and our new chunk.\n-                lines.append(self.buffer[:-1] + \"\\n\")\n-                self.buffer = \"\"\n-                text = text[1:]\n-            else:\n-                # Handle the case where we have \"\\r\" at the end of our\n-                # previous input.\n-                lines.append(self.buffer[:-1] + \"\\n\")\n-                self.buffer = \"\"\n-\n-        while text:\n-            num_chars = len(text)\n-            for idx in range(num_chars):\n-                char = text[idx]\n-                next_char = None if idx + 1 == num_chars else text[idx + 1]\n-                if char == \"\\n\":\n-                    lines.append(self.buffer + text[: idx + 1])\n-                    self.buffer = \"\"\n-                    text = text[idx + 1 :]\n-                    break\n-                elif char == \"\\r\" and next_char == \"\\n\":\n-                    lines.append(self.buffer + text[:idx] + \"\\n\")\n-                    self.buffer = \"\"\n-                    text = text[idx + 2 :]\n-                    break\n-                elif char == \"\\r\" and next_char is not None:\n-                    lines.append(self.buffer + text[:idx] + \"\\n\")\n-                    self.buffer = \"\"\n-                    text = text[idx + 1 :]\n-                    break\n-                elif next_char is None:\n-                    self.buffer += text\n-                    text = \"\"\n-                    break\n+        # See https://docs.python.org/3/library/stdtypes.html#str.splitlines\n+        NEWLINE_CHARS = \"\\n\\r\\x0b\\x0c\\x1c\\x1d\\x1e\\x85\\u2028\\u2029\"\n+\n+        # We always push a trailing `\\r` into the next decode iteration.\n+        if self.trailing_cr:\n+            text = \"\\r\" + text\n+            self.trailing_cr = False\n+        if text.endswith(\"\\r\"):\n+            self.trailing_cr = True\n+            text = text[:-1]\n+\n+        if not text:\n+            return []\n+\n+        trailing_newline = text[-1] in NEWLINE_CHARS\n+        lines = text.splitlines()\n+\n+        if len(lines) == 1 and not trailing_newline:\n+            # No new lines, buffer the input and continue.\n+            self.buffer.append(lines[0])\n+            return []\n+\n+        if self.buffer:\n+            # Include any existing buffer in the first portion of the\n+            # splitlines result.\n+            lines = [\"\".join(self.buffer) + lines[0]] + lines[1:]\n+            self.buffer = []\n+\n+        if not trailing_newline:\n+            # If the last segment of splitlines is not newline terminated,\n+            # then drop it from our output and start a new buffer.\n+            self.buffer = [lines.pop()]\n \n         return lines\n \n     def flush(self) -> typing.List[str]:\n-        if self.buffer.endswith(\"\\r\"):\n-            # Handle the case where we had a trailing '\\r', which could have\n-            # been a '\\r\\n' pair.\n-            lines = [self.buffer[:-1] + \"\\n\"]\n-        elif self.buffer:\n-            lines = [self.buffer]\n-        else:\n-            lines = []\n-        self.buffer = \"\"\n+        if not self.buffer and not self.trailing_cr:\n+            return []\n+\n+        lines = [\"\".join(self.buffer)]\n+        self.buffer = []\n+        self.trailing_cr = False\n         return lines\n", "issue": "LineDecoder is accidentally quadratic: iter_lines() seems to hang forever\nWhen calling `Response.iter_lines()`, things can seem to hang forever.\r\n\r\nThe problem is that `LineDecoder` is quadratic in it's string copying behaviour. If a 31MB chunk with 18,768 lines is passed in to `LineDecoder()` then it takes 1m45s to process it vs 0.1s for a simple `text.splitlines()`. This is readily reproduced by simply using `LineDecoder()` to decode a file, like `/usr/share/dict/words` (do not attempt, you may be waiting until the heat death of the universe).\r\n\r\nIt may be a bug somewhere else that `iter_text()` is returning chunks that are too big.\r\n\r\nBut either way, you probably want to reconsider the wisdom of slicing the string to chop the beginning off it vs. just keeping an index.\r\n\r\nYou can slice the string before returning from the decoder so it's only done once per chunk.\r\n\r\nAnyhoo, thanks for the great work!\n", "before_files": [{"content": "\"\"\"\nHandlers for Content-Encoding.\n\nSee: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Encoding\n\"\"\"\nimport codecs\nimport io\nimport typing\nimport zlib\n\nfrom ._compat import brotli\nfrom ._exceptions import DecodingError\n\n\nclass ContentDecoder:\n    def decode(self, data: bytes) -> bytes:\n        raise NotImplementedError()  # pragma: no cover\n\n    def flush(self) -> bytes:\n        raise NotImplementedError()  # pragma: no cover\n\n\nclass IdentityDecoder(ContentDecoder):\n    \"\"\"\n    Handle unencoded data.\n    \"\"\"\n\n    def decode(self, data: bytes) -> bytes:\n        return data\n\n    def flush(self) -> bytes:\n        return b\"\"\n\n\nclass DeflateDecoder(ContentDecoder):\n    \"\"\"\n    Handle 'deflate' decoding.\n\n    See: https://stackoverflow.com/questions/1838699\n    \"\"\"\n\n    def __init__(self) -> None:\n        self.first_attempt = True\n        self.decompressor = zlib.decompressobj()\n\n    def decode(self, data: bytes) -> bytes:\n        was_first_attempt = self.first_attempt\n        self.first_attempt = False\n        try:\n            return self.decompressor.decompress(data)\n        except zlib.error as exc:\n            if was_first_attempt:\n                self.decompressor = zlib.decompressobj(-zlib.MAX_WBITS)\n                return self.decode(data)\n            raise DecodingError(str(exc)) from exc\n\n    def flush(self) -> bytes:\n        try:\n            return self.decompressor.flush()\n        except zlib.error as exc:  # pragma: no cover\n            raise DecodingError(str(exc)) from exc\n\n\nclass GZipDecoder(ContentDecoder):\n    \"\"\"\n    Handle 'gzip' decoding.\n\n    See: https://stackoverflow.com/questions/1838699\n    \"\"\"\n\n    def __init__(self) -> None:\n        self.decompressor = zlib.decompressobj(zlib.MAX_WBITS | 16)\n\n    def decode(self, data: bytes) -> bytes:\n        try:\n            return self.decompressor.decompress(data)\n        except zlib.error as exc:\n            raise DecodingError(str(exc)) from exc\n\n    def flush(self) -> bytes:\n        try:\n            return self.decompressor.flush()\n        except zlib.error as exc:  # pragma: no cover\n            raise DecodingError(str(exc)) from exc\n\n\nclass BrotliDecoder(ContentDecoder):\n    \"\"\"\n    Handle 'brotli' decoding.\n\n    Requires `pip install brotlipy`. See: https://brotlipy.readthedocs.io/\n        or   `pip install brotli`. See https://github.com/google/brotli\n    Supports both 'brotlipy' and 'Brotli' packages since they share an import\n    name. The top branches are for 'brotlipy' and bottom branches for 'Brotli'\n    \"\"\"\n\n    def __init__(self) -> None:\n        if brotli is None:  # pragma: no cover\n            raise ImportError(\n                \"Using 'BrotliDecoder', but neither of the 'brotlicffi' or 'brotli' \"\n                \"packages have been installed. \"\n                \"Make sure to install httpx using `pip install httpx[brotli]`.\"\n            ) from None\n\n        self.decompressor = brotli.Decompressor()\n        self.seen_data = False\n        self._decompress: typing.Callable[[bytes], bytes]\n        if hasattr(self.decompressor, \"decompress\"):\n            # The 'brotlicffi' package.\n            self._decompress = self.decompressor.decompress  # pragma: no cover\n        else:\n            # The 'brotli' package.\n            self._decompress = self.decompressor.process  # pragma: no cover\n\n    def decode(self, data: bytes) -> bytes:\n        if not data:\n            return b\"\"\n        self.seen_data = True\n        try:\n            return self._decompress(data)\n        except brotli.error as exc:\n            raise DecodingError(str(exc)) from exc\n\n    def flush(self) -> bytes:\n        if not self.seen_data:\n            return b\"\"\n        try:\n            if hasattr(self.decompressor, \"finish\"):\n                # Only available in the 'brotlicffi' package.\n\n                # As the decompressor decompresses eagerly, this\n                # will never actually emit any data. However, it will potentially throw\n                # errors if a truncated or damaged data stream has been used.\n                self.decompressor.finish()  # pragma: no cover\n            return b\"\"\n        except brotli.error as exc:  # pragma: no cover\n            raise DecodingError(str(exc)) from exc\n\n\nclass MultiDecoder(ContentDecoder):\n    \"\"\"\n    Handle the case where multiple encodings have been applied.\n    \"\"\"\n\n    def __init__(self, children: typing.Sequence[ContentDecoder]) -> None:\n        \"\"\"\n        'children' should be a sequence of decoders in the order in which\n        each was applied.\n        \"\"\"\n        # Note that we reverse the order for decoding.\n        self.children = list(reversed(children))\n\n    def decode(self, data: bytes) -> bytes:\n        for child in self.children:\n            data = child.decode(data)\n        return data\n\n    def flush(self) -> bytes:\n        data = b\"\"\n        for child in self.children:\n            data = child.decode(data) + child.flush()\n        return data\n\n\nclass ByteChunker:\n    \"\"\"\n    Handles returning byte content in fixed-size chunks.\n    \"\"\"\n\n    def __init__(self, chunk_size: typing.Optional[int] = None) -> None:\n        self._buffer = io.BytesIO()\n        self._chunk_size = chunk_size\n\n    def decode(self, content: bytes) -> typing.List[bytes]:\n        if self._chunk_size is None:\n            return [content] if content else []\n\n        self._buffer.write(content)\n        if self._buffer.tell() >= self._chunk_size:\n            value = self._buffer.getvalue()\n            chunks = [\n                value[i : i + self._chunk_size]\n                for i in range(0, len(value), self._chunk_size)\n            ]\n            if len(chunks[-1]) == self._chunk_size:\n                self._buffer.seek(0)\n                self._buffer.truncate()\n                return chunks\n            else:\n                self._buffer.seek(0)\n                self._buffer.write(chunks[-1])\n                self._buffer.truncate()\n                return chunks[:-1]\n        else:\n            return []\n\n    def flush(self) -> typing.List[bytes]:\n        value = self._buffer.getvalue()\n        self._buffer.seek(0)\n        self._buffer.truncate()\n        return [value] if value else []\n\n\nclass TextChunker:\n    \"\"\"\n    Handles returning text content in fixed-size chunks.\n    \"\"\"\n\n    def __init__(self, chunk_size: typing.Optional[int] = None) -> None:\n        self._buffer = io.StringIO()\n        self._chunk_size = chunk_size\n\n    def decode(self, content: str) -> typing.List[str]:\n        if self._chunk_size is None:\n            return [content]\n\n        self._buffer.write(content)\n        if self._buffer.tell() >= self._chunk_size:\n            value = self._buffer.getvalue()\n            chunks = [\n                value[i : i + self._chunk_size]\n                for i in range(0, len(value), self._chunk_size)\n            ]\n            if len(chunks[-1]) == self._chunk_size:\n                self._buffer.seek(0)\n                self._buffer.truncate()\n                return chunks\n            else:\n                self._buffer.seek(0)\n                self._buffer.write(chunks[-1])\n                self._buffer.truncate()\n                return chunks[:-1]\n        else:\n            return []\n\n    def flush(self) -> typing.List[str]:\n        value = self._buffer.getvalue()\n        self._buffer.seek(0)\n        self._buffer.truncate()\n        return [value] if value else []\n\n\nclass TextDecoder:\n    \"\"\"\n    Handles incrementally decoding bytes into text\n    \"\"\"\n\n    def __init__(self, encoding: str = \"utf-8\"):\n        self.decoder = codecs.getincrementaldecoder(encoding)(errors=\"replace\")\n\n    def decode(self, data: bytes) -> str:\n        return self.decoder.decode(data)\n\n    def flush(self) -> str:\n        return self.decoder.decode(b\"\", True)\n\n\nclass LineDecoder:\n    \"\"\"\n    Handles incrementally reading lines from text.\n\n    Uses universal line decoding, supporting any of `\\n`, `\\r`, or `\\r\\n`\n    as line endings, normalizing to `\\n`.\n    \"\"\"\n\n    def __init__(self) -> None:\n        self.buffer = \"\"\n\n    def decode(self, text: str) -> typing.List[str]:\n        lines = []\n\n        if text and self.buffer and self.buffer[-1] == \"\\r\":\n            if text.startswith(\"\\n\"):\n                # Handle the case where we have an \"\\r\\n\" split across\n                # our previous input, and our new chunk.\n                lines.append(self.buffer[:-1] + \"\\n\")\n                self.buffer = \"\"\n                text = text[1:]\n            else:\n                # Handle the case where we have \"\\r\" at the end of our\n                # previous input.\n                lines.append(self.buffer[:-1] + \"\\n\")\n                self.buffer = \"\"\n\n        while text:\n            num_chars = len(text)\n            for idx in range(num_chars):\n                char = text[idx]\n                next_char = None if idx + 1 == num_chars else text[idx + 1]\n                if char == \"\\n\":\n                    lines.append(self.buffer + text[: idx + 1])\n                    self.buffer = \"\"\n                    text = text[idx + 1 :]\n                    break\n                elif char == \"\\r\" and next_char == \"\\n\":\n                    lines.append(self.buffer + text[:idx] + \"\\n\")\n                    self.buffer = \"\"\n                    text = text[idx + 2 :]\n                    break\n                elif char == \"\\r\" and next_char is not None:\n                    lines.append(self.buffer + text[:idx] + \"\\n\")\n                    self.buffer = \"\"\n                    text = text[idx + 1 :]\n                    break\n                elif next_char is None:\n                    self.buffer += text\n                    text = \"\"\n                    break\n\n        return lines\n\n    def flush(self) -> typing.List[str]:\n        if self.buffer.endswith(\"\\r\"):\n            # Handle the case where we had a trailing '\\r', which could have\n            # been a '\\r\\n' pair.\n            lines = [self.buffer[:-1] + \"\\n\"]\n        elif self.buffer:\n            lines = [self.buffer]\n        else:\n            lines = []\n        self.buffer = \"\"\n        return lines\n\n\nSUPPORTED_DECODERS = {\n    \"identity\": IdentityDecoder,\n    \"gzip\": GZipDecoder,\n    \"deflate\": DeflateDecoder,\n    \"br\": BrotliDecoder,\n}\n\n\nif brotli is None:\n    SUPPORTED_DECODERS.pop(\"br\")  # pragma: no cover\n", "path": "httpx/_decoders.py"}], "after_files": [{"content": "\"\"\"\nHandlers for Content-Encoding.\n\nSee: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Encoding\n\"\"\"\nimport codecs\nimport io\nimport typing\nimport zlib\n\nfrom ._compat import brotli\nfrom ._exceptions import DecodingError\n\n\nclass ContentDecoder:\n    def decode(self, data: bytes) -> bytes:\n        raise NotImplementedError()  # pragma: no cover\n\n    def flush(self) -> bytes:\n        raise NotImplementedError()  # pragma: no cover\n\n\nclass IdentityDecoder(ContentDecoder):\n    \"\"\"\n    Handle unencoded data.\n    \"\"\"\n\n    def decode(self, data: bytes) -> bytes:\n        return data\n\n    def flush(self) -> bytes:\n        return b\"\"\n\n\nclass DeflateDecoder(ContentDecoder):\n    \"\"\"\n    Handle 'deflate' decoding.\n\n    See: https://stackoverflow.com/questions/1838699\n    \"\"\"\n\n    def __init__(self) -> None:\n        self.first_attempt = True\n        self.decompressor = zlib.decompressobj()\n\n    def decode(self, data: bytes) -> bytes:\n        was_first_attempt = self.first_attempt\n        self.first_attempt = False\n        try:\n            return self.decompressor.decompress(data)\n        except zlib.error as exc:\n            if was_first_attempt:\n                self.decompressor = zlib.decompressobj(-zlib.MAX_WBITS)\n                return self.decode(data)\n            raise DecodingError(str(exc)) from exc\n\n    def flush(self) -> bytes:\n        try:\n            return self.decompressor.flush()\n        except zlib.error as exc:  # pragma: no cover\n            raise DecodingError(str(exc)) from exc\n\n\nclass GZipDecoder(ContentDecoder):\n    \"\"\"\n    Handle 'gzip' decoding.\n\n    See: https://stackoverflow.com/questions/1838699\n    \"\"\"\n\n    def __init__(self) -> None:\n        self.decompressor = zlib.decompressobj(zlib.MAX_WBITS | 16)\n\n    def decode(self, data: bytes) -> bytes:\n        try:\n            return self.decompressor.decompress(data)\n        except zlib.error as exc:\n            raise DecodingError(str(exc)) from exc\n\n    def flush(self) -> bytes:\n        try:\n            return self.decompressor.flush()\n        except zlib.error as exc:  # pragma: no cover\n            raise DecodingError(str(exc)) from exc\n\n\nclass BrotliDecoder(ContentDecoder):\n    \"\"\"\n    Handle 'brotli' decoding.\n\n    Requires `pip install brotlipy`. See: https://brotlipy.readthedocs.io/\n        or   `pip install brotli`. See https://github.com/google/brotli\n    Supports both 'brotlipy' and 'Brotli' packages since they share an import\n    name. The top branches are for 'brotlipy' and bottom branches for 'Brotli'\n    \"\"\"\n\n    def __init__(self) -> None:\n        if brotli is None:  # pragma: no cover\n            raise ImportError(\n                \"Using 'BrotliDecoder', but neither of the 'brotlicffi' or 'brotli' \"\n                \"packages have been installed. \"\n                \"Make sure to install httpx using `pip install httpx[brotli]`.\"\n            ) from None\n\n        self.decompressor = brotli.Decompressor()\n        self.seen_data = False\n        self._decompress: typing.Callable[[bytes], bytes]\n        if hasattr(self.decompressor, \"decompress\"):\n            # The 'brotlicffi' package.\n            self._decompress = self.decompressor.decompress  # pragma: no cover\n        else:\n            # The 'brotli' package.\n            self._decompress = self.decompressor.process  # pragma: no cover\n\n    def decode(self, data: bytes) -> bytes:\n        if not data:\n            return b\"\"\n        self.seen_data = True\n        try:\n            return self._decompress(data)\n        except brotli.error as exc:\n            raise DecodingError(str(exc)) from exc\n\n    def flush(self) -> bytes:\n        if not self.seen_data:\n            return b\"\"\n        try:\n            if hasattr(self.decompressor, \"finish\"):\n                # Only available in the 'brotlicffi' package.\n\n                # As the decompressor decompresses eagerly, this\n                # will never actually emit any data. However, it will potentially throw\n                # errors if a truncated or damaged data stream has been used.\n                self.decompressor.finish()  # pragma: no cover\n            return b\"\"\n        except brotli.error as exc:  # pragma: no cover\n            raise DecodingError(str(exc)) from exc\n\n\nclass MultiDecoder(ContentDecoder):\n    \"\"\"\n    Handle the case where multiple encodings have been applied.\n    \"\"\"\n\n    def __init__(self, children: typing.Sequence[ContentDecoder]) -> None:\n        \"\"\"\n        'children' should be a sequence of decoders in the order in which\n        each was applied.\n        \"\"\"\n        # Note that we reverse the order for decoding.\n        self.children = list(reversed(children))\n\n    def decode(self, data: bytes) -> bytes:\n        for child in self.children:\n            data = child.decode(data)\n        return data\n\n    def flush(self) -> bytes:\n        data = b\"\"\n        for child in self.children:\n            data = child.decode(data) + child.flush()\n        return data\n\n\nclass ByteChunker:\n    \"\"\"\n    Handles returning byte content in fixed-size chunks.\n    \"\"\"\n\n    def __init__(self, chunk_size: typing.Optional[int] = None) -> None:\n        self._buffer = io.BytesIO()\n        self._chunk_size = chunk_size\n\n    def decode(self, content: bytes) -> typing.List[bytes]:\n        if self._chunk_size is None:\n            return [content] if content else []\n\n        self._buffer.write(content)\n        if self._buffer.tell() >= self._chunk_size:\n            value = self._buffer.getvalue()\n            chunks = [\n                value[i : i + self._chunk_size]\n                for i in range(0, len(value), self._chunk_size)\n            ]\n            if len(chunks[-1]) == self._chunk_size:\n                self._buffer.seek(0)\n                self._buffer.truncate()\n                return chunks\n            else:\n                self._buffer.seek(0)\n                self._buffer.write(chunks[-1])\n                self._buffer.truncate()\n                return chunks[:-1]\n        else:\n            return []\n\n    def flush(self) -> typing.List[bytes]:\n        value = self._buffer.getvalue()\n        self._buffer.seek(0)\n        self._buffer.truncate()\n        return [value] if value else []\n\n\nclass TextChunker:\n    \"\"\"\n    Handles returning text content in fixed-size chunks.\n    \"\"\"\n\n    def __init__(self, chunk_size: typing.Optional[int] = None) -> None:\n        self._buffer = io.StringIO()\n        self._chunk_size = chunk_size\n\n    def decode(self, content: str) -> typing.List[str]:\n        if self._chunk_size is None:\n            return [content]\n\n        self._buffer.write(content)\n        if self._buffer.tell() >= self._chunk_size:\n            value = self._buffer.getvalue()\n            chunks = [\n                value[i : i + self._chunk_size]\n                for i in range(0, len(value), self._chunk_size)\n            ]\n            if len(chunks[-1]) == self._chunk_size:\n                self._buffer.seek(0)\n                self._buffer.truncate()\n                return chunks\n            else:\n                self._buffer.seek(0)\n                self._buffer.write(chunks[-1])\n                self._buffer.truncate()\n                return chunks[:-1]\n        else:\n            return []\n\n    def flush(self) -> typing.List[str]:\n        value = self._buffer.getvalue()\n        self._buffer.seek(0)\n        self._buffer.truncate()\n        return [value] if value else []\n\n\nclass TextDecoder:\n    \"\"\"\n    Handles incrementally decoding bytes into text\n    \"\"\"\n\n    def __init__(self, encoding: str = \"utf-8\"):\n        self.decoder = codecs.getincrementaldecoder(encoding)(errors=\"replace\")\n\n    def decode(self, data: bytes) -> str:\n        return self.decoder.decode(data)\n\n    def flush(self) -> str:\n        return self.decoder.decode(b\"\", True)\n\n\nclass LineDecoder:\n    \"\"\"\n    Handles incrementally reading lines from text.\n\n    Has the same behaviour as the stdllib splitlines, but handling the input iteratively.\n    \"\"\"\n\n    def __init__(self) -> None:\n        self.buffer: typing.List[str] = []\n        self.trailing_cr: bool = False\n\n    def decode(self, text: str) -> typing.List[str]:\n        # See https://docs.python.org/3/library/stdtypes.html#str.splitlines\n        NEWLINE_CHARS = \"\\n\\r\\x0b\\x0c\\x1c\\x1d\\x1e\\x85\\u2028\\u2029\"\n\n        # We always push a trailing `\\r` into the next decode iteration.\n        if self.trailing_cr:\n            text = \"\\r\" + text\n            self.trailing_cr = False\n        if text.endswith(\"\\r\"):\n            self.trailing_cr = True\n            text = text[:-1]\n\n        if not text:\n            return []\n\n        trailing_newline = text[-1] in NEWLINE_CHARS\n        lines = text.splitlines()\n\n        if len(lines) == 1 and not trailing_newline:\n            # No new lines, buffer the input and continue.\n            self.buffer.append(lines[0])\n            return []\n\n        if self.buffer:\n            # Include any existing buffer in the first portion of the\n            # splitlines result.\n            lines = [\"\".join(self.buffer) + lines[0]] + lines[1:]\n            self.buffer = []\n\n        if not trailing_newline:\n            # If the last segment of splitlines is not newline terminated,\n            # then drop it from our output and start a new buffer.\n            self.buffer = [lines.pop()]\n\n        return lines\n\n    def flush(self) -> typing.List[str]:\n        if not self.buffer and not self.trailing_cr:\n            return []\n\n        lines = [\"\".join(self.buffer)]\n        self.buffer = []\n        self.trailing_cr = False\n        return lines\n\n\nSUPPORTED_DECODERS = {\n    \"identity\": IdentityDecoder,\n    \"gzip\": GZipDecoder,\n    \"deflate\": DeflateDecoder,\n    \"br\": BrotliDecoder,\n}\n\n\nif brotli is None:\n    SUPPORTED_DECODERS.pop(\"br\")  # pragma: no cover\n", "path": "httpx/_decoders.py"}]}