We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
StreamResponse write() does not raise an exception if client disconnects
### Describe the bug

Hi!

I have an application that uses a `StreamResponse` to continuously write data to the client by calling the `write()` method. The server stops sending data when the client disconnects. In version 3.8.1 the `write()` method raises an exception when the client disconnects, but in later version it does not. The commit that introduces the bug is 20c93656d18b50c17c85b7a1e234276fb1444f27.

I've tested on Python 3.7, 3.8 and 3.10.

### To Reproduce

Add later if really needed.

### Expected behavior

An exception is raised when the client disconnects.

### Logs/tracebacks

```python-traceback
-
```


### Python Version

```console
$ python --version
Python 3.10.7
```


### aiohttp Version

```console
$ python -m pip show aiohttp
Name: aiohttp
Version: 3.8.3
...
```


### multidict Version

```console
-
```


### yarl Version

```console
-
```


### OS

Linux

### Related component

Server

### Additional context

_No response_

### Code of Conduct

- [X] I agree to follow the aio-libs Code of Conduct

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `aiohttp/base_protocol.py`
Content:
```
1 import asyncio
2 from typing import Optional, cast
3 
4 from .tcp_helpers import tcp_nodelay
5 
6 
7 class BaseProtocol(asyncio.Protocol):
8     __slots__ = (
9         "_loop",
10         "_paused",
11         "_drain_waiter",
12         "_connection_lost",
13         "_reading_paused",
14         "transport",
15     )
16 
17     def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
18         self._loop: asyncio.AbstractEventLoop = loop
19         self._paused = False
20         self._drain_waiter: Optional[asyncio.Future[None]] = None
21         self._connection_lost = False
22         self._reading_paused = False
23 
24         self.transport: Optional[asyncio.Transport] = None
25 
26     def pause_writing(self) -> None:
27         assert not self._paused
28         self._paused = True
29 
30     def resume_writing(self) -> None:
31         assert self._paused
32         self._paused = False
33 
34         waiter = self._drain_waiter
35         if waiter is not None:
36             self._drain_waiter = None
37             if not waiter.done():
38                 waiter.set_result(None)
39 
40     def pause_reading(self) -> None:
41         if not self._reading_paused and self.transport is not None:
42             try:
43                 self.transport.pause_reading()
44             except (AttributeError, NotImplementedError, RuntimeError):
45                 pass
46             self._reading_paused = True
47 
48     def resume_reading(self) -> None:
49         if self._reading_paused and self.transport is not None:
50             try:
51                 self.transport.resume_reading()
52             except (AttributeError, NotImplementedError, RuntimeError):
53                 pass
54             self._reading_paused = False
55 
56     def connection_made(self, transport: asyncio.BaseTransport) -> None:
57         tr = cast(asyncio.Transport, transport)
58         tcp_nodelay(tr, True)
59         self.transport = tr
60 
61     def connection_lost(self, exc: Optional[BaseException]) -> None:
62         self._connection_lost = True
63         # Wake up the writer if currently paused.
64         self.transport = None
65         if not self._paused:
66             return
67         waiter = self._drain_waiter
68         if waiter is None:
69             return
70         self._drain_waiter = None
71         if waiter.done():
72             return
73         if exc is None:
74             waiter.set_result(None)
75         else:
76             waiter.set_exception(exc)
77 
78     async def _drain_helper(self) -> None:
79         if self._connection_lost:
80             raise ConnectionResetError("Connection lost")
81         if not self._paused:
82             return
83         waiter = self._drain_waiter
84         if waiter is None:
85             waiter = self._loop.create_future()
86             self._drain_waiter = waiter
87         await asyncio.shield(waiter)
88
```

Path: `aiohttp/http_writer.py`
Content:
```
1 """Http related parsers and protocol."""
2 
3 import asyncio
4 import zlib
5 from typing import Any, Awaitable, Callable, NamedTuple, Optional, Union  # noqa
6 
7 from multidict import CIMultiDict
8 
9 from .abc import AbstractStreamWriter
10 from .base_protocol import BaseProtocol
11 from .helpers import NO_EXTENSIONS
12 
13 __all__ = ("StreamWriter", "HttpVersion", "HttpVersion10", "HttpVersion11")
14 
15 
16 class HttpVersion(NamedTuple):
17     major: int
18     minor: int
19 
20 
21 HttpVersion10 = HttpVersion(1, 0)
22 HttpVersion11 = HttpVersion(1, 1)
23 
24 
25 _T_OnChunkSent = Optional[Callable[[bytes], Awaitable[None]]]
26 _T_OnHeadersSent = Optional[Callable[["CIMultiDict[str]"], Awaitable[None]]]
27 
28 
29 class StreamWriter(AbstractStreamWriter):
30     def __init__(
31         self,
32         protocol: BaseProtocol,
33         loop: asyncio.AbstractEventLoop,
34         on_chunk_sent: _T_OnChunkSent = None,
35         on_headers_sent: _T_OnHeadersSent = None,
36     ) -> None:
37         self._protocol = protocol
38         self._transport = protocol.transport
39 
40         self.loop = loop
41         self.length = None
42         self.chunked = False
43         self.buffer_size = 0
44         self.output_size = 0
45 
46         self._eof = False
47         self._compress: Any = None
48         self._drain_waiter = None
49 
50         self._on_chunk_sent: _T_OnChunkSent = on_chunk_sent
51         self._on_headers_sent: _T_OnHeadersSent = on_headers_sent
52 
53     @property
54     def transport(self) -> Optional[asyncio.Transport]:
55         return self._transport
56 
57     @property
58     def protocol(self) -> BaseProtocol:
59         return self._protocol
60 
61     def enable_chunking(self) -> None:
62         self.chunked = True
63 
64     def enable_compression(
65         self, encoding: str = "deflate", strategy: int = zlib.Z_DEFAULT_STRATEGY
66     ) -> None:
67         zlib_mode = 16 + zlib.MAX_WBITS if encoding == "gzip" else zlib.MAX_WBITS
68         self._compress = zlib.compressobj(wbits=zlib_mode, strategy=strategy)
69 
70     def _write(self, chunk: bytes) -> None:
71         size = len(chunk)
72         self.buffer_size += size
73         self.output_size += size
74 
75         if self._transport is None or self._transport.is_closing():
76             raise ConnectionResetError("Cannot write to closing transport")
77         self._transport.write(chunk)
78 
79     async def write(
80         self, chunk: bytes, *, drain: bool = True, LIMIT: int = 0x10000
81     ) -> None:
82         """Writes chunk of data to a stream.
83 
84         write_eof() indicates end of stream.
85         writer can't be used after write_eof() method being called.
86         write() return drain future.
87         """
88         if self._on_chunk_sent is not None:
89             await self._on_chunk_sent(chunk)
90 
91         if isinstance(chunk, memoryview):
92             if chunk.nbytes != len(chunk):
93                 # just reshape it
94                 chunk = chunk.cast("c")
95 
96         if self._compress is not None:
97             chunk = self._compress.compress(chunk)
98             if not chunk:
99                 return
100 
101         if self.length is not None:
102             chunk_len = len(chunk)
103             if self.length >= chunk_len:
104                 self.length = self.length - chunk_len
105             else:
106                 chunk = chunk[: self.length]
107                 self.length = 0
108                 if not chunk:
109                     return
110 
111         if chunk:
112             if self.chunked:
113                 chunk_len_pre = ("%x\r\n" % len(chunk)).encode("ascii")
114                 chunk = chunk_len_pre + chunk + b"\r\n"
115 
116             self._write(chunk)
117 
118             if self.buffer_size > LIMIT and drain:
119                 self.buffer_size = 0
120                 await self.drain()
121 
122     async def write_headers(
123         self, status_line: str, headers: "CIMultiDict[str]"
124     ) -> None:
125         """Write request/response status and headers."""
126         if self._on_headers_sent is not None:
127             await self._on_headers_sent(headers)
128 
129         # status + headers
130         buf = _serialize_headers(status_line, headers)
131         self._write(buf)
132 
133     async def write_eof(self, chunk: bytes = b"") -> None:
134         if self._eof:
135             return
136 
137         if chunk and self._on_chunk_sent is not None:
138             await self._on_chunk_sent(chunk)
139 
140         if self._compress:
141             if chunk:
142                 chunk = self._compress.compress(chunk)
143 
144             chunk = chunk + self._compress.flush()
145             if chunk and self.chunked:
146                 chunk_len = ("%x\r\n" % len(chunk)).encode("ascii")
147                 chunk = chunk_len + chunk + b"\r\n0\r\n\r\n"
148         else:
149             if self.chunked:
150                 if chunk:
151                     chunk_len = ("%x\r\n" % len(chunk)).encode("ascii")
152                     chunk = chunk_len + chunk + b"\r\n0\r\n\r\n"
153                 else:
154                     chunk = b"0\r\n\r\n"
155 
156         if chunk:
157             self._write(chunk)
158 
159         await self.drain()
160 
161         self._eof = True
162         self._transport = None
163 
164     async def drain(self) -> None:
165         """Flush the write buffer.
166 
167         The intended use is to write
168 
169           await w.write(data)
170           await w.drain()
171         """
172         if self._protocol.transport is not None:
173             await self._protocol._drain_helper()
174 
175 
176 def _safe_header(string: str) -> str:
177     if "\r" in string or "\n" in string:
178         raise ValueError(
179             "Newline or carriage return detected in headers. "
180             "Potential header injection attack."
181         )
182     return string
183 
184 
185 def _py_serialize_headers(status_line: str, headers: "CIMultiDict[str]") -> bytes:
186     headers_gen = (_safe_header(k) + ": " + _safe_header(v) for k, v in headers.items())
187     line = status_line + "\r\n" + "\r\n".join(headers_gen) + "\r\n\r\n"
188     return line.encode("utf-8")
189 
190 
191 _serialize_headers = _py_serialize_headers
192 
193 try:
194     import aiohttp._http_writer as _http_writer  # type: ignore[import]
195 
196     _c_serialize_headers = _http_writer._serialize_headers
197     if not NO_EXTENSIONS:
198         _serialize_headers = _c_serialize_headers
199 except ImportError:
200     pass
201
```
--- 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/aiohttp/base_protocol.py b/aiohttp/base_protocol.py
--- a/aiohttp/base_protocol.py
+++ b/aiohttp/base_protocol.py
@@ -18,11 +18,15 @@
         self._loop: asyncio.AbstractEventLoop = loop
         self._paused = False
         self._drain_waiter: Optional[asyncio.Future[None]] = None
-        self._connection_lost = False
         self._reading_paused = False
 
         self.transport: Optional[asyncio.Transport] = None
 
+    @property
+    def connected(self) -> bool:
+        """Return True if the connection is open."""
+        return self.transport is not None
+
     def pause_writing(self) -> None:
         assert not self._paused
         self._paused = True
@@ -59,7 +63,6 @@
         self.transport = tr
 
     def connection_lost(self, exc: Optional[BaseException]) -> None:
-        self._connection_lost = True
         # Wake up the writer if currently paused.
         self.transport = None
         if not self._paused:
@@ -76,7 +79,7 @@
             waiter.set_exception(exc)
 
     async def _drain_helper(self) -> None:
-        if self._connection_lost:
+        if not self.connected:
             raise ConnectionResetError("Connection lost")
         if not self._paused:
             return
diff --git a/aiohttp/http_writer.py b/aiohttp/http_writer.py
--- a/aiohttp/http_writer.py
+++ b/aiohttp/http_writer.py
@@ -35,7 +35,6 @@
         on_headers_sent: _T_OnHeadersSent = None,
     ) -> None:
         self._protocol = protocol
-        self._transport = protocol.transport
 
         self.loop = loop
         self.length = None
@@ -52,7 +51,7 @@
 
     @property
     def transport(self) -> Optional[asyncio.Transport]:
-        return self._transport
+        return self._protocol.transport
 
     @property
     def protocol(self) -> BaseProtocol:
@@ -71,10 +70,10 @@
         size = len(chunk)
         self.buffer_size += size
         self.output_size += size
-
-        if self._transport is None or self._transport.is_closing():
+        transport = self.transport
+        if not self._protocol.connected or transport is None or transport.is_closing():
             raise ConnectionResetError("Cannot write to closing transport")
-        self._transport.write(chunk)
+        transport.write(chunk)
 
     async def write(
         self, chunk: bytes, *, drain: bool = True, LIMIT: int = 0x10000
@@ -159,7 +158,6 @@
         await self.drain()
 
         self._eof = True
-        self._transport = None
 
     async def drain(self) -> None:
         """Flush the write buffer.

{"golden_diff": "diff --git a/aiohttp/base_protocol.py b/aiohttp/base_protocol.py\n--- a/aiohttp/base_protocol.py\n+++ b/aiohttp/base_protocol.py\n@@ -18,11 +18,15 @@\n         self._loop: asyncio.AbstractEventLoop = loop\n         self._paused = False\n         self._drain_waiter: Optional[asyncio.Future[None]] = None\n-        self._connection_lost = False\n         self._reading_paused = False\n \n         self.transport: Optional[asyncio.Transport] = None\n \n+    @property\n+    def connected(self) -> bool:\n+        \"\"\"Return True if the connection is open.\"\"\"\n+        return self.transport is not None\n+\n     def pause_writing(self) -> None:\n         assert not self._paused\n         self._paused = True\n@@ -59,7 +63,6 @@\n         self.transport = tr\n \n     def connection_lost(self, exc: Optional[BaseException]) -> None:\n-        self._connection_lost = True\n         # Wake up the writer if currently paused.\n         self.transport = None\n         if not self._paused:\n@@ -76,7 +79,7 @@\n             waiter.set_exception(exc)\n \n     async def _drain_helper(self) -> None:\n-        if self._connection_lost:\n+        if not self.connected:\n             raise ConnectionResetError(\"Connection lost\")\n         if not self._paused:\n             return\ndiff --git a/aiohttp/http_writer.py b/aiohttp/http_writer.py\n--- a/aiohttp/http_writer.py\n+++ b/aiohttp/http_writer.py\n@@ -35,7 +35,6 @@\n         on_headers_sent: _T_OnHeadersSent = None,\n     ) -> None:\n         self._protocol = protocol\n-        self._transport = protocol.transport\n \n         self.loop = loop\n         self.length = None\n@@ -52,7 +51,7 @@\n \n     @property\n     def transport(self) -> Optional[asyncio.Transport]:\n-        return self._transport\n+        return self._protocol.transport\n \n     @property\n     def protocol(self) -> BaseProtocol:\n@@ -71,10 +70,10 @@\n         size = len(chunk)\n         self.buffer_size += size\n         self.output_size += size\n-\n-        if self._transport is None or self._transport.is_closing():\n+        transport = self.transport\n+        if not self._protocol.connected or transport is None or transport.is_closing():\n             raise ConnectionResetError(\"Cannot write to closing transport\")\n-        self._transport.write(chunk)\n+        transport.write(chunk)\n \n     async def write(\n         self, chunk: bytes, *, drain: bool = True, LIMIT: int = 0x10000\n@@ -159,7 +158,6 @@\n         await self.drain()\n \n         self._eof = True\n-        self._transport = None\n \n     async def drain(self) -> None:\n         \"\"\"Flush the write buffer.\n", "issue": "StreamResponse write() does not raise an exception if client disconnects\n### Describe the bug\n\nHi!\r\n\r\nI have an application that uses a `StreamResponse` to continuously write data to the client by calling the `write()` method. The server stops sending data when the client disconnects. In version 3.8.1 the `write()` method raises an exception when the client disconnects, but in later version it does not. The commit that introduces the bug is 20c93656d18b50c17c85b7a1e234276fb1444f27.\r\n\r\nI've tested on Python 3.7, 3.8 and 3.10.\n\n### To Reproduce\n\nAdd later if really needed.\n\n### Expected behavior\n\nAn exception is raised when the client disconnects.\n\n### Logs/tracebacks\n\n```python-traceback\n-\n```\n\n\n### Python Version\n\n```console\n$ python --version\r\nPython 3.10.7\n```\n\n\n### aiohttp Version\n\n```console\n$ python -m pip show aiohttp\r\nName: aiohttp\r\nVersion: 3.8.3\r\n...\n```\n\n\n### multidict Version\n\n```console\n-\n```\n\n\n### yarl Version\n\n```console\n-\n```\n\n\n### OS\n\nLinux\n\n### Related component\n\nServer\n\n### Additional context\n\n_No response_\n\n### Code of Conduct\n\n- [X] I agree to follow the aio-libs Code of Conduct\n", "before_files": [{"content": "import asyncio\nfrom typing import Optional, cast\n\nfrom .tcp_helpers import tcp_nodelay\n\n\nclass BaseProtocol(asyncio.Protocol):\n    __slots__ = (\n        \"_loop\",\n        \"_paused\",\n        \"_drain_waiter\",\n        \"_connection_lost\",\n        \"_reading_paused\",\n        \"transport\",\n    )\n\n    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:\n        self._loop: asyncio.AbstractEventLoop = loop\n        self._paused = False\n        self._drain_waiter: Optional[asyncio.Future[None]] = None\n        self._connection_lost = False\n        self._reading_paused = False\n\n        self.transport: Optional[asyncio.Transport] = None\n\n    def pause_writing(self) -> None:\n        assert not self._paused\n        self._paused = True\n\n    def resume_writing(self) -> None:\n        assert self._paused\n        self._paused = False\n\n        waiter = self._drain_waiter\n        if waiter is not None:\n            self._drain_waiter = None\n            if not waiter.done():\n                waiter.set_result(None)\n\n    def pause_reading(self) -> None:\n        if not self._reading_paused and self.transport is not None:\n            try:\n                self.transport.pause_reading()\n            except (AttributeError, NotImplementedError, RuntimeError):\n                pass\n            self._reading_paused = True\n\n    def resume_reading(self) -> None:\n        if self._reading_paused and self.transport is not None:\n            try:\n                self.transport.resume_reading()\n            except (AttributeError, NotImplementedError, RuntimeError):\n                pass\n            self._reading_paused = False\n\n    def connection_made(self, transport: asyncio.BaseTransport) -> None:\n        tr = cast(asyncio.Transport, transport)\n        tcp_nodelay(tr, True)\n        self.transport = tr\n\n    def connection_lost(self, exc: Optional[BaseException]) -> None:\n        self._connection_lost = True\n        # Wake up the writer if currently paused.\n        self.transport = None\n        if not self._paused:\n            return\n        waiter = self._drain_waiter\n        if waiter is None:\n            return\n        self._drain_waiter = None\n        if waiter.done():\n            return\n        if exc is None:\n            waiter.set_result(None)\n        else:\n            waiter.set_exception(exc)\n\n    async def _drain_helper(self) -> None:\n        if self._connection_lost:\n            raise ConnectionResetError(\"Connection lost\")\n        if not self._paused:\n            return\n        waiter = self._drain_waiter\n        if waiter is None:\n            waiter = self._loop.create_future()\n            self._drain_waiter = waiter\n        await asyncio.shield(waiter)\n", "path": "aiohttp/base_protocol.py"}, {"content": "\"\"\"Http related parsers and protocol.\"\"\"\n\nimport asyncio\nimport zlib\nfrom typing import Any, Awaitable, Callable, NamedTuple, Optional, Union  # noqa\n\nfrom multidict import CIMultiDict\n\nfrom .abc import AbstractStreamWriter\nfrom .base_protocol import BaseProtocol\nfrom .helpers import NO_EXTENSIONS\n\n__all__ = (\"StreamWriter\", \"HttpVersion\", \"HttpVersion10\", \"HttpVersion11\")\n\n\nclass HttpVersion(NamedTuple):\n    major: int\n    minor: int\n\n\nHttpVersion10 = HttpVersion(1, 0)\nHttpVersion11 = HttpVersion(1, 1)\n\n\n_T_OnChunkSent = Optional[Callable[[bytes], Awaitable[None]]]\n_T_OnHeadersSent = Optional[Callable[[\"CIMultiDict[str]\"], Awaitable[None]]]\n\n\nclass StreamWriter(AbstractStreamWriter):\n    def __init__(\n        self,\n        protocol: BaseProtocol,\n        loop: asyncio.AbstractEventLoop,\n        on_chunk_sent: _T_OnChunkSent = None,\n        on_headers_sent: _T_OnHeadersSent = None,\n    ) -> None:\n        self._protocol = protocol\n        self._transport = protocol.transport\n\n        self.loop = loop\n        self.length = None\n        self.chunked = False\n        self.buffer_size = 0\n        self.output_size = 0\n\n        self._eof = False\n        self._compress: Any = None\n        self._drain_waiter = None\n\n        self._on_chunk_sent: _T_OnChunkSent = on_chunk_sent\n        self._on_headers_sent: _T_OnHeadersSent = on_headers_sent\n\n    @property\n    def transport(self) -> Optional[asyncio.Transport]:\n        return self._transport\n\n    @property\n    def protocol(self) -> BaseProtocol:\n        return self._protocol\n\n    def enable_chunking(self) -> None:\n        self.chunked = True\n\n    def enable_compression(\n        self, encoding: str = \"deflate\", strategy: int = zlib.Z_DEFAULT_STRATEGY\n    ) -> None:\n        zlib_mode = 16 + zlib.MAX_WBITS if encoding == \"gzip\" else zlib.MAX_WBITS\n        self._compress = zlib.compressobj(wbits=zlib_mode, strategy=strategy)\n\n    def _write(self, chunk: bytes) -> None:\n        size = len(chunk)\n        self.buffer_size += size\n        self.output_size += size\n\n        if self._transport is None or self._transport.is_closing():\n            raise ConnectionResetError(\"Cannot write to closing transport\")\n        self._transport.write(chunk)\n\n    async def write(\n        self, chunk: bytes, *, drain: bool = True, LIMIT: int = 0x10000\n    ) -> None:\n        \"\"\"Writes chunk of data to a stream.\n\n        write_eof() indicates end of stream.\n        writer can't be used after write_eof() method being called.\n        write() return drain future.\n        \"\"\"\n        if self._on_chunk_sent is not None:\n            await self._on_chunk_sent(chunk)\n\n        if isinstance(chunk, memoryview):\n            if chunk.nbytes != len(chunk):\n                # just reshape it\n                chunk = chunk.cast(\"c\")\n\n        if self._compress is not None:\n            chunk = self._compress.compress(chunk)\n            if not chunk:\n                return\n\n        if self.length is not None:\n            chunk_len = len(chunk)\n            if self.length >= chunk_len:\n                self.length = self.length - chunk_len\n            else:\n                chunk = chunk[: self.length]\n                self.length = 0\n                if not chunk:\n                    return\n\n        if chunk:\n            if self.chunked:\n                chunk_len_pre = (\"%x\\r\\n\" % len(chunk)).encode(\"ascii\")\n                chunk = chunk_len_pre + chunk + b\"\\r\\n\"\n\n            self._write(chunk)\n\n            if self.buffer_size > LIMIT and drain:\n                self.buffer_size = 0\n                await self.drain()\n\n    async def write_headers(\n        self, status_line: str, headers: \"CIMultiDict[str]\"\n    ) -> None:\n        \"\"\"Write request/response status and headers.\"\"\"\n        if self._on_headers_sent is not None:\n            await self._on_headers_sent(headers)\n\n        # status + headers\n        buf = _serialize_headers(status_line, headers)\n        self._write(buf)\n\n    async def write_eof(self, chunk: bytes = b\"\") -> None:\n        if self._eof:\n            return\n\n        if chunk and self._on_chunk_sent is not None:\n            await self._on_chunk_sent(chunk)\n\n        if self._compress:\n            if chunk:\n                chunk = self._compress.compress(chunk)\n\n            chunk = chunk + self._compress.flush()\n            if chunk and self.chunked:\n                chunk_len = (\"%x\\r\\n\" % len(chunk)).encode(\"ascii\")\n                chunk = chunk_len + chunk + b\"\\r\\n0\\r\\n\\r\\n\"\n        else:\n            if self.chunked:\n                if chunk:\n                    chunk_len = (\"%x\\r\\n\" % len(chunk)).encode(\"ascii\")\n                    chunk = chunk_len + chunk + b\"\\r\\n0\\r\\n\\r\\n\"\n                else:\n                    chunk = b\"0\\r\\n\\r\\n\"\n\n        if chunk:\n            self._write(chunk)\n\n        await self.drain()\n\n        self._eof = True\n        self._transport = None\n\n    async def drain(self) -> None:\n        \"\"\"Flush the write buffer.\n\n        The intended use is to write\n\n          await w.write(data)\n          await w.drain()\n        \"\"\"\n        if self._protocol.transport is not None:\n            await self._protocol._drain_helper()\n\n\ndef _safe_header(string: str) -> str:\n    if \"\\r\" in string or \"\\n\" in string:\n        raise ValueError(\n            \"Newline or carriage return detected in headers. \"\n            \"Potential header injection attack.\"\n        )\n    return string\n\n\ndef _py_serialize_headers(status_line: str, headers: \"CIMultiDict[str]\") -> bytes:\n    headers_gen = (_safe_header(k) + \": \" + _safe_header(v) for k, v in headers.items())\n    line = status_line + \"\\r\\n\" + \"\\r\\n\".join(headers_gen) + \"\\r\\n\\r\\n\"\n    return line.encode(\"utf-8\")\n\n\n_serialize_headers = _py_serialize_headers\n\ntry:\n    import aiohttp._http_writer as _http_writer  # type: ignore[import]\n\n    _c_serialize_headers = _http_writer._serialize_headers\n    if not NO_EXTENSIONS:\n        _serialize_headers = _c_serialize_headers\nexcept ImportError:\n    pass\n", "path": "aiohttp/http_writer.py"}], "after_files": [{"content": "import asyncio\nfrom typing import Optional, cast\n\nfrom .tcp_helpers import tcp_nodelay\n\n\nclass BaseProtocol(asyncio.Protocol):\n    __slots__ = (\n        \"_loop\",\n        \"_paused\",\n        \"_drain_waiter\",\n        \"_connection_lost\",\n        \"_reading_paused\",\n        \"transport\",\n    )\n\n    def __init__(self, loop: asyncio.AbstractEventLoop) -> None:\n        self._loop: asyncio.AbstractEventLoop = loop\n        self._paused = False\n        self._drain_waiter: Optional[asyncio.Future[None]] = None\n        self._reading_paused = False\n\n        self.transport: Optional[asyncio.Transport] = None\n\n    @property\n    def connected(self) -> bool:\n        \"\"\"Return True if the connection is open.\"\"\"\n        return self.transport is not None\n\n    def pause_writing(self) -> None:\n        assert not self._paused\n        self._paused = True\n\n    def resume_writing(self) -> None:\n        assert self._paused\n        self._paused = False\n\n        waiter = self._drain_waiter\n        if waiter is not None:\n            self._drain_waiter = None\n            if not waiter.done():\n                waiter.set_result(None)\n\n    def pause_reading(self) -> None:\n        if not self._reading_paused and self.transport is not None:\n            try:\n                self.transport.pause_reading()\n            except (AttributeError, NotImplementedError, RuntimeError):\n                pass\n            self._reading_paused = True\n\n    def resume_reading(self) -> None:\n        if self._reading_paused and self.transport is not None:\n            try:\n                self.transport.resume_reading()\n            except (AttributeError, NotImplementedError, RuntimeError):\n                pass\n            self._reading_paused = False\n\n    def connection_made(self, transport: asyncio.BaseTransport) -> None:\n        tr = cast(asyncio.Transport, transport)\n        tcp_nodelay(tr, True)\n        self.transport = tr\n\n    def connection_lost(self, exc: Optional[BaseException]) -> None:\n        # Wake up the writer if currently paused.\n        self.transport = None\n        if not self._paused:\n            return\n        waiter = self._drain_waiter\n        if waiter is None:\n            return\n        self._drain_waiter = None\n        if waiter.done():\n            return\n        if exc is None:\n            waiter.set_result(None)\n        else:\n            waiter.set_exception(exc)\n\n    async def _drain_helper(self) -> None:\n        if not self.connected:\n            raise ConnectionResetError(\"Connection lost\")\n        if not self._paused:\n            return\n        waiter = self._drain_waiter\n        if waiter is None:\n            waiter = self._loop.create_future()\n            self._drain_waiter = waiter\n        await asyncio.shield(waiter)\n", "path": "aiohttp/base_protocol.py"}, {"content": "\"\"\"Http related parsers and protocol.\"\"\"\n\nimport asyncio\nimport zlib\nfrom typing import Any, Awaitable, Callable, NamedTuple, Optional, Union  # noqa\n\nfrom multidict import CIMultiDict\n\nfrom .abc import AbstractStreamWriter\nfrom .base_protocol import BaseProtocol\nfrom .helpers import NO_EXTENSIONS\n\n__all__ = (\"StreamWriter\", \"HttpVersion\", \"HttpVersion10\", \"HttpVersion11\")\n\n\nclass HttpVersion(NamedTuple):\n    major: int\n    minor: int\n\n\nHttpVersion10 = HttpVersion(1, 0)\nHttpVersion11 = HttpVersion(1, 1)\n\n\n_T_OnChunkSent = Optional[Callable[[bytes], Awaitable[None]]]\n_T_OnHeadersSent = Optional[Callable[[\"CIMultiDict[str]\"], Awaitable[None]]]\n\n\nclass StreamWriter(AbstractStreamWriter):\n    def __init__(\n        self,\n        protocol: BaseProtocol,\n        loop: asyncio.AbstractEventLoop,\n        on_chunk_sent: _T_OnChunkSent = None,\n        on_headers_sent: _T_OnHeadersSent = None,\n    ) -> None:\n        self._protocol = protocol\n\n        self.loop = loop\n        self.length = None\n        self.chunked = False\n        self.buffer_size = 0\n        self.output_size = 0\n\n        self._eof = False\n        self._compress: Any = None\n        self._drain_waiter = None\n\n        self._on_chunk_sent: _T_OnChunkSent = on_chunk_sent\n        self._on_headers_sent: _T_OnHeadersSent = on_headers_sent\n\n    @property\n    def transport(self) -> Optional[asyncio.Transport]:\n        return self._protocol.transport\n\n    @property\n    def protocol(self) -> BaseProtocol:\n        return self._protocol\n\n    def enable_chunking(self) -> None:\n        self.chunked = True\n\n    def enable_compression(\n        self, encoding: str = \"deflate\", strategy: int = zlib.Z_DEFAULT_STRATEGY\n    ) -> None:\n        zlib_mode = 16 + zlib.MAX_WBITS if encoding == \"gzip\" else zlib.MAX_WBITS\n        self._compress = zlib.compressobj(wbits=zlib_mode, strategy=strategy)\n\n    def _write(self, chunk: bytes) -> None:\n        size = len(chunk)\n        self.buffer_size += size\n        self.output_size += size\n        transport = self.transport\n        if not self._protocol.connected or transport is None or transport.is_closing():\n            raise ConnectionResetError(\"Cannot write to closing transport\")\n        transport.write(chunk)\n\n    async def write(\n        self, chunk: bytes, *, drain: bool = True, LIMIT: int = 0x10000\n    ) -> None:\n        \"\"\"Writes chunk of data to a stream.\n\n        write_eof() indicates end of stream.\n        writer can't be used after write_eof() method being called.\n        write() return drain future.\n        \"\"\"\n        if self._on_chunk_sent is not None:\n            await self._on_chunk_sent(chunk)\n\n        if isinstance(chunk, memoryview):\n            if chunk.nbytes != len(chunk):\n                # just reshape it\n                chunk = chunk.cast(\"c\")\n\n        if self._compress is not None:\n            chunk = self._compress.compress(chunk)\n            if not chunk:\n                return\n\n        if self.length is not None:\n            chunk_len = len(chunk)\n            if self.length >= chunk_len:\n                self.length = self.length - chunk_len\n            else:\n                chunk = chunk[: self.length]\n                self.length = 0\n                if not chunk:\n                    return\n\n        if chunk:\n            if self.chunked:\n                chunk_len_pre = (\"%x\\r\\n\" % len(chunk)).encode(\"ascii\")\n                chunk = chunk_len_pre + chunk + b\"\\r\\n\"\n\n            self._write(chunk)\n\n            if self.buffer_size > LIMIT and drain:\n                self.buffer_size = 0\n                await self.drain()\n\n    async def write_headers(\n        self, status_line: str, headers: \"CIMultiDict[str]\"\n    ) -> None:\n        \"\"\"Write request/response status and headers.\"\"\"\n        if self._on_headers_sent is not None:\n            await self._on_headers_sent(headers)\n\n        # status + headers\n        buf = _serialize_headers(status_line, headers)\n        self._write(buf)\n\n    async def write_eof(self, chunk: bytes = b\"\") -> None:\n        if self._eof:\n            return\n\n        if chunk and self._on_chunk_sent is not None:\n            await self._on_chunk_sent(chunk)\n\n        if self._compress:\n            if chunk:\n                chunk = self._compress.compress(chunk)\n\n            chunk = chunk + self._compress.flush()\n            if chunk and self.chunked:\n                chunk_len = (\"%x\\r\\n\" % len(chunk)).encode(\"ascii\")\n                chunk = chunk_len + chunk + b\"\\r\\n0\\r\\n\\r\\n\"\n        else:\n            if self.chunked:\n                if chunk:\n                    chunk_len = (\"%x\\r\\n\" % len(chunk)).encode(\"ascii\")\n                    chunk = chunk_len + chunk + b\"\\r\\n0\\r\\n\\r\\n\"\n                else:\n                    chunk = b\"0\\r\\n\\r\\n\"\n\n        if chunk:\n            self._write(chunk)\n\n        await self.drain()\n\n        self._eof = True\n\n    async def drain(self) -> None:\n        \"\"\"Flush the write buffer.\n\n        The intended use is to write\n\n          await w.write(data)\n          await w.drain()\n        \"\"\"\n        if self._protocol.transport is not None:\n            await self._protocol._drain_helper()\n\n\ndef _safe_header(string: str) -> str:\n    if \"\\r\" in string or \"\\n\" in string:\n        raise ValueError(\n            \"Newline or carriage return detected in headers. \"\n            \"Potential header injection attack.\"\n        )\n    return string\n\n\ndef _py_serialize_headers(status_line: str, headers: \"CIMultiDict[str]\") -> bytes:\n    headers_gen = (_safe_header(k) + \": \" + _safe_header(v) for k, v in headers.items())\n    line = status_line + \"\\r\\n\" + \"\\r\\n\".join(headers_gen) + \"\\r\\n\\r\\n\"\n    return line.encode(\"utf-8\")\n\n\n_serialize_headers = _py_serialize_headers\n\ntry:\n    import aiohttp._http_writer as _http_writer  # type: ignore[import]\n\n    _c_serialize_headers = _http_writer._serialize_headers\n    if not NO_EXTENSIONS:\n        _serialize_headers = _c_serialize_headers\nexcept ImportError:\n    pass\n", "path": "aiohttp/http_writer.py"}]}