Class Connection

src/websockets/sync/connection.py:32–1078

:mod:`threading` implementation of a WebSocket connection. :class:`Connection` provides APIs shared between WebSocket servers and clients. You shouldn't use it directly. Instead, use :class:`~websockets.sync.client.ClientConnection` or :class:`~websockets.sync.server.Serve

class Connection:
    """
    :mod:`threading` implementation of a WebSocket connection.

    :class:`Connection` provides APIs shared between WebSocket servers and
    clients.

    You shouldn't use it directly. Instead, use
    :class:`~websockets.sync.client.ClientConnection` or
    :class:`~websockets.sync.server.ServerConnection`.

    """

    recv_bufsize = 65536

    def __init__(
        self,
        socket: socket.socket,
        protocol: Protocol,
        *,
        ping_interval: float | None = 20,
        ping_timeout: float | None = 20,
        close_timeout: float | None = 10,
        max_queue: int | None | tuple[int | None, int | None] = 16,
    ) -> None:
        self.socket = socket
        self.protocol = protocol
        self.ping_interval = ping_interval
        self.ping_timeout = ping_timeout
        self.close_timeout = close_timeout
        if isinstance(max_queue, int) or max_queue is None:
            max_queue_high, max_queue_low = max_queue, None
        else:
            max_queue_high, max_queue_low = max_queue

        # Inject reference to this instance in the protocol's logger.
        self.protocol.logger = logging.LoggerAdapter(
            self.protocol.logger,
            {"websocket": self},
        )

        # Copy attributes from the protocol for convenience.
        self.id: uuid.UUID = self.protocol.id
        """Unique identifier of the connection. Useful in logs."""
        self.logger: LoggerLike = self.protocol.logger
        """Logger for this connection."""
        self.debug = self.protocol.debug

        # HTTP handshake request and response.
        self.request: Request | None = None
        """Opening handshake request."""
        self.response: Response | None = None
        """Opening handshake response."""

        # Mutex serializing interactions with the protocol.
        self.protocol_mutex = threading.Lock()

        # Lock stopping reads when the assembler buffer is full.