1from __future__ import annotations
 2
 3import asyncio
 4from typing import TYPE_CHECKING
 5
 6from . import util
 7
 8if TYPE_CHECKING:
 9    from .app import ServerApplication
10
11# Keep-alive connection timeout in seconds
12KEEPALIVE = 2
13
14
15class Connection:
16    def __init__(
17        self,
18        app: ServerApplication,
19        reader: asyncio.StreamReader,
20        writer: asyncio.StreamWriter,
21        client: tuple[str, int],
22        server: tuple[str, int],
23        *,
24        is_ssl: bool = False,
25    ) -> None:
26        self.app = app
27        self.reader = reader
28        self.writer = writer
29        self.client = client
30        self.server = server
31
32        self.is_h2: bool = False
33        self.is_ssl: bool = is_ssl
34        self.req_count: int = 0
35
36        # Byte read during keepalive wait, prepended to next header read
37        self._keepalive_byte: bytes = b""
38
39    def close(self) -> None:
40        if not self.writer.is_closing():
41            self.writer.close()
42
43    async def recv(self, n: int) -> bytes:
44        """Read up to n bytes from a connection."""
45        return await self.reader.read(n)
46
47    async def sendall(self, data: bytes) -> None:
48        """Send all bytes on a connection."""
49        self.writer.write(data)
50        await self.writer.drain()
51
52    async def write_error(self, status_int: int, reason: str, mesg: str) -> None:
53        """Send an HTTP error response on a connection."""
54        await self.sendall(util._error_response_bytes(status_int, reason, mesg))
55
56    async def wait_readable(self) -> None:
57        data = await self.reader.read(1)
58        if data:
59            # Prepend the peeked byte back into the buffer
60            self._keepalive_byte = data