Class Protocol

src/websockets/protocol.py:71–768

Sans-I/O implementation of a WebSocket connection. Args: side: :attr:`~Side.CLIENT` or :attr:`~Side.SERVER`. state: Initial state of the WebSocket connection. max_size: Maximum size of incoming messages in bytes. :obj:`None` disables the limit. You may p

class Protocol:
    """
    Sans-I/O implementation of a WebSocket connection.

    Args:
        side: :attr:`~Side.CLIENT` or :attr:`~Side.SERVER`.
        state: Initial state of the WebSocket connection.
        max_size: Maximum size of incoming messages in bytes.
            :obj:`None` disables the limit. You may pass a ``(max_message_size,
            max_fragment_size)`` tuple to set different limits for messages and
            fragments when you expect long messages sent in short fragments.
        logger: Logger for this connection; depending on ``side``,
            defaults to ``logging.getLogger("websockets.client")``
            or ``logging.getLogger("websockets.server")``;
            see the :doc:`logging guide <../../topics/logging>` for details.

    """

    def __init__(
        self,
        side: Side,
        *,
        state: State = OPEN,
        max_size: tuple[int | None, int | None] | int | None = 2**20,
        logger: LoggerLike | None = None,
    ) -> None:
        # Unique identifier. For logs.
        self.id: uuid.UUID = uuid.uuid4()
        """Unique identifier of the connection. Useful in logs."""

        # Logger or LoggerAdapter for this connection.
        if logger is None:
            logger = logging.getLogger(f"websockets.{side.name.lower()}")
        self.logger: LoggerLike = logger
        """Logger for this connection."""

        # Track if DEBUG is enabled. Shortcut logging calls if it isn't.
        self.debug = logger.isEnabledFor(logging.DEBUG)

        # Connection side. CLIENT or SERVER.
        self.side = side

        # Connection state. Initially OPEN because subclasses handle CONNECTING.
        self.state = state

        # Maximum size of incoming messages in bytes.
        if isinstance(max_size, int) or max_size is None:
            self.max_message_size, self.max_fragment_size = max_size, None
        else:
            self.max_message_size, self.max_fragment_size = max_size

        # Current size of incoming message in bytes. Only set while reading a
        # fragmented message i.e. a data frames with the FIN bit not set.
        self.current_size: int | None = None

        # True while sending a fragmented message i.e. a data frames with the
        # FIN bit not set.
        self.expect_continuation_frame = False