content: from __future__ import annotations
  
  import asyncio
  import codecs
  import collections
  import logging
  import random
  import ssl
  import struct
  import sys
  import time
  import traceback
  import uuid
  import warnings
  from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Iterable, Mapping
  from typing import Any, Callable, Deque, cast
  
  from ..asyncio.compatibility import asyncio_timeout
  from ..datastructures import Headers
  from ..exceptions import (
      ConnectionClosed,
      ConnectionClosedError,
      ConnectionClosedOK,
      InvalidState,
      PayloadTooBig,
      ProtocolError,
  )
  from ..extensions import Extension
  from ..frames import (
      OK_CLOSE_CODES,
      OP_BINARY,
      OP_CLOSE,
      OP_CONT,
      OP_PING,
      OP_PONG,
      OP_TEXT,
      Close,
      CloseCode,
      Opcode,
  )
  from ..protocol import State
  from ..typing import BytesLike, Data, DataLike, LoggerLike, Subprotocol
  from .framing import Frame, prepare_ctrl, prepare_data
  
  
  __all__ = ["WebSocketCommonProtocol"]
  
  
  # In order to ensure consistency, the code always checks the current value of
  # WebSocketCommonProtocol.state before assigning a new value and never yields
  # between the check and the assignment.
  
  
  class WebSocketCommonProtocol(asyncio.Protocol):
      """
      WebSocket connection.
  
      :class:`WebSocketCommonProtocol` provides APIs shared between WebSocket
      servers and clients. You shouldn't use it directly. Instead, use
      :class:`~websockets.legacy.client.WebSocketClientProtocol` or
      :class:`~websockets.legacy.server.WebSocketServerProtocol`.
  
      This documentation focuses on low-level details that aren't covered in the
      documentation of :class:`~websockets.legacy.client.WebSocketClientProtocol`
      and :class:`~websockets.legacy.server.WebSocketServerProtocol` for the sake
      of simplicity.
  
      Once the connection is open, a Ping_ frame is sent every ``ping_interval``
      seconds. This serves as a keepalive. It helps keeping the connection open,
      especially in the presence of proxies with short timeouts on inactive
      connections. Set ``ping_interval`` to :obj:`None` to disable this behavior.
  
      .. _Ping: https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.2
  
      If the corresponding Pong_ frame isn't received within ``ping_timeout``
      seconds, the connection is considered unusable and is closed with code 1011.
      This ensures that the remote endpoint remains responsive. Set
      ``ping_timeout`` to :obj:`None` to disable this behavior.
  
      .. _Pong: https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.3
  
      See the discussion of :doc:`keepalive <../../topics/keepalive>` for details.
  
      The ``close_timeout`` parameter defines a maximum wait time for completing
      the closing handshake and terminating the TCP connection. For legacy
      reasons, :meth:`close` completes in at most ``5 * close_timeout`` seconds
      for clients and ``4 * close_timeout`` for servers.
  
      ``close_timeout`` is a parameter of the protocol because websockets usually
      calls :meth:`close` implicitly upon exit:
  
      * on the client side, when using :func:`~websockets.legacy.client.connect`
        as a context manager;
      * on the server side, when the connection handler terminates.
  
      To apply a timeout to any other API, wrap it in :func:`~asyncio.timeout` or
      :func:`~asyncio.wait_for`.
  
      The ``max_size`` parameter enforces the maximum size for incoming messages
      in bytes. The default value is 1 MiB. If a larger message is received,
      :meth:`recv` will raise :exc:`~websockets.exceptions.ConnectionClosedError`
      and the connection will be closed with code 1009.
  
      The ``max_queue`` parameter sets the maximum length of the queue that
      holds incoming messages. The default value is ``32``. Messages are added
      to an in-memory queue when they're received; then :meth:`recv` pops from
      that queue. In order to prevent excessive memory consumption when
      messages are received faster than they can be processed, the queue must
      be bounded. If the queue fills up, the protocol stops processing incoming
      data until :meth:`recv` is called. In this situation, various receive
      buffers (at least in :mod:`asyncio` and in the OS) will fill up, then the
      TCP receive window will shrink, slowing down transmission to avoid packet
      loss.
  
      Since Python can use up to 4 bytes of memory to represent a single
      character, each connection may use up to ``4 * max_size * max_queue``
      bytes of memory to store incoming messages. By default, this is 128 MiB.
      You may want to lower the limits, depending on your application's
      requirements.
  
      The ``read_limit`` argument sets the high-water limit of the buffer for
      incoming bytes. The low-water limit is half the high-water limit. The
      default value is 64 KiB, half of asyncio's default (based on the current
      implementation of :class:`~asyncio.StreamReader`).
  
      The ``write_limit`` argument sets the high-water limit of the buffer for
      outgoing bytes. The low-water limit is a quarter of the high-water limit.
      The default value is 64 KiB, equal to asyncio's default (based on the
      current implementation of ``FlowControlMixin``).
  
      See the discussion of :doc:`memory usage <../../topics/memory>` for details.
  
      Args:
          logger: Logger for this server.
              It defaults to ``logging.getLogger("websockets.protocol")``.
              See the :doc:`logging guide <../../topics/logging>` for details.
          ping_interval: Interval between keepalive pings in seconds.
              :obj:`None` disables keepalive.
          ping_timeout: Timeout for keepalive pings in seconds.
              :obj:`None` disables timeouts.
          close_timeout: Timeout for closing the connection in seconds.
              For legacy reasons, the actual timeout is 4 or 5 times larger.
          max_size: Maximum size of incoming messages in bytes.
              :obj:`None` disables the limit.
          max_queue: Maximum number of incoming messages in receive buffer.
              :obj:`None` disables the limit.
          read_limit: High-water mark of read buffer in bytes.
          write_limit: High-water mark of write buffer in bytes.
  
      """
  
      # There are only two differences between the client-side and server-side
      # behavior: masking the payload and closing the underlying TCP connection.
      # Set is_client = True/False and side = "client"/"server" to pick a side.
      is_client: bool
      side: str = "undefined"
  
      def __init__(
          self,
          *,
          logger: LoggerLike | None = None,
          ping_interval: float | None = 20,
          ping_timeout: float | None = 20,
          close_timeout: float | None = None,
          max_size: int | None = 2**20,
          max_queue: int | None = 2**5,
          read_limit: int = 2**16,
          write_limit: int = 2**16,
          # The following arguments are kept only for backwards compatibility.
          host: str | None = None,
          port: int | None = None,
          secure: bool | None = None,
          legacy_recv: bool = False,
          loop: asyncio.AbstractEventLoop | None = None,
          timeout: float | None = None,
      ) -> None:
          if legacy_recv:  # pragma: no cover
              warnings.warn("legacy_recv is deprecated", DeprecationWarning)
  
          # Backwards compatibility: close_timeout used to be called timeout.
          if timeout is None:
              timeout = 10
          else:
              warnings.warn("rename timeout to close_timeout", DeprecationWarning)
          # If both are specified, timeout is ignored.
          if close_timeout is None:
              close_timeout = timeout
  
          # Backwards compatibility: the loop parameter used to be supported.
          if loop is None:
              loop = asyncio.get_event_loop()
          else:
              warnings.warn("remove loop argument", DeprecationWarning)
  
          self.ping_interval = ping_interval
          self.ping_timeout = ping_timeout
          self.close_timeout = close_timeout
          self.max_size = max_size
          self.max_queue = max_queue
          self.read_limit = read_limit
          self.write_limit = write_limit
  
          # 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("websockets.protocol")
          self.logger: LoggerLike = logging.LoggerAdapter(logger, {"websocket": self})
          """Logger for this connection."""
  
          # Track if DEBUG is enabled. Shortcut logging calls if it isn't.
          self.debug = logger.isEnabledFor(logging.DEBUG)
  
          self.loop = loop
  
          self._host = host
          self._port = port
          self._secure = secure
          self.legacy_recv = legacy_recv
  
          # Configure read buffer limits. The high-water limit is defined by
          # ``self.read_limit``. The ``limit`` argument controls the line length
          # limit and half the buffer limit of :class:`~asyncio.StreamReader`.
          # That's why it must be set to half of ``self.read_limit``.
          self.reader = asyncio.StreamReader(limit=read_limit // 2, loop=loop)
  
          # Copied from asyncio.FlowControlMixin
          self._paused = False
          self._drain_waiter: asyncio.Future[None] | None = None
  
          # This class implements the data transfer and closing handshake, which
          # are shared between the client-side and the server-side.
          # Subclasses implement the opening handshake and, on success, execute
          # :meth:`connection_open` to change the state to OPEN.
          self.state = State.CONNECTING
          if self.debug:
              self.logger.debug("= connection is CONNECTING")
  
          # HTTP protocol parameters.
          self.path: str
          """Path of the opening handshake request."""
          self.request_headers: Headers
          """Opening handshake request headers."""
          self.response_headers: Headers
          """Opening handshake response headers."""
  
          # WebSocket protocol parameters.
          self.extensions: list[Extension] = []
          self.subprotocol: Subprotocol | None = None
          """Subprotocol, if one was negotiated."""
  
          # Close code and reason, set when a close frame is sent or received.
          self.close_rcvd: Close | None = None
          self.close_sent: Close | None = None
          self.close_rcvd_then_sent: bool | None = None
  
          # Completed when the connection state becomes CLOSED. Translates the
          # :meth:`connection_lost` callback to a :class:`~asyncio.Future`
          # that can be awaited. (Other :class:`~asyncio.Protocol` callbacks are
          # translated by ``self.stream_reader``).
          self.connection_lost_waiter: asyncio.Future[None] = loop.create_future()
  
          # Queue of received messages.
          self.messages: Deque[Data] = collections.deque()
          self._pop_message_waiter: asyncio.Future[None] | None = None
          self._put_message_waiter: asyncio.Future[None] | None = None
  
          # Protect sending fragmented messages.
          self._fragmented_message_waiter: asyncio.Future[None] | None = None
  
          # Mapping of ping IDs to pong waiters, in chronological order.
          self.pings: dict[bytes, tuple[asyncio.Future[float], float]] = {}
  
          self.latency: float = 0
          """
          Latency of the connection, in seconds.
  
          Latency is defined as the round-trip time of the connection. It is
          measured by sending a Ping frame and waiting for a matching Pong frame.
          Before the first measurement, :attr:`latency` is ``0``.
  
          By default, websockets enables a :ref:`keepalive <keepalive>` mechanism
          that sends Ping frames automatically at regular intervals. You can also
          send Ping frames and measure latency with :meth:`ping`.
          """
  
          # Task running the data transfer.
          self.transfer_data_task: asyncio.Task[None]
  
          # Exception that occurred during data transfer, if any.
          self.transfer_data_exc: BaseException | None = None
  
          # Task sending keepalive pings.
          self.keepalive_ping_task: asyncio.Task[None]
  
          # Task closing the TCP connection.
          self.close_connection_task: asyncio.Task[None]
  
      # Copied from asyncio.FlowControlMixin
      async def _drain_helper(self) -> None:  # pragma: no cover
          if self.connection_lost_waiter.done():
              raise ConnectionResetError("Connection lost")
          if not self._paused:
              return
          waiter = self._drain_waiter
          assert waiter is None or waiter.cancelled()
          waiter = self.loop.create_future()
          self._drain_waiter = waiter
          await waiter
  
      # Copied from asyncio.StreamWriter
      async def _drain(self) -> None:  # pragma: no cover
          if self.reader is not None:
              exc = self.reader.exception()
              if exc is not None:
                  raise exc
          if self.transport is not None:
              if self.transport.is_closing():
                  # Yield to the event loop so connection_lost() may be
                  # called.  Without this, _drain_helper() would return
                  # immediately, and code that calls
                  #     write(...); yield from drain()
                  # in a loop would never call connection_lost(), so it
                  # would not see an error when the socket is closed.
                  await asyncio.sleep(0)
          await self._drain_helper()
  
      def connection_open(self) -> None:
          """
          Callback when the WebSocket opening handshake completes.
  
          Enter the OPEN state and start the data transfer phase.
  
          """
          # 4.1. The WebSocket Connection is Established.
          assert self.state is State.CONNECTING
          self.state = State.OPEN
          if self.debug:
              self.logger.debug("= connection is OPEN")
          # Start the task that receives incoming WebSocket messages.
          self.transfer_data_task = self.loop.create_task(self.transfer_data())
          # Start the task that sends pings at regular intervals.
          self.keepalive_ping_task = self.loop.create_task(self.keepalive_ping())
          # Start the task that eventually closes the TCP connection.
          self.close_connection_task = self.loop.create_task(self.close_connection())
  
      @property
      def host(self) -> str | None:
          alternative = "remote_address" if self.is_client else "local_address"
          warnings.warn(f"use {alternative}[0] instead of host", DeprecationWarning)
          return self._host
  
      @property
      def port(self) -> int | None:
          alternative = "remote_address" if self.is_client else "local_address"
          warnings.warn(f"use {alternative}[1] instead of port", DeprecationWarning)
          return self._port
  
      @property
      def secure(self) -> bool | None:
          warnings.warn("don't use secure", DeprecationWarning)
          return self._secure
  
      # Public API
  
      @property
      def local_address(self) -> Any:
          """
          Local address of the connection.
  
          For IPv4 connections, this is a ``(host, port)`` tuple.
  
          The format of the address depends on the address family;
          see :meth:`~socket.socket.getsockname`.
  
          :obj:`None` if the TCP connection isn't established yet.
  
          """
          try:
              transport = self.transport
          except AttributeError:
              return None
          else:
              return transport.get_extra_info("sockname")
  
      @property
      def remote_address(self) -> Any:
          """
          Remote address of the connection.
  
          For IPv4 connections, this is a ``(host, port)`` tuple.
  
          The format of the address depends on the address family;
          see :meth:`~socket.socket.getpeername`.
  
          :obj:`None` if the TCP connection isn't established yet.
  
          """
          try:
              transport = self.transport
          except AttributeError:
              return None
          else:
              return transport.get_extra_info("peername")
  
      @property
      def open(self) -> bool:
          """
          :obj:`True` when the connection is open; :obj:`False` otherwise.
  
          This attribute may be used to detect disconnections. However, this
          approach is discouraged per the EAFP_ principle. Instead, you should
          handle :exc:`~websockets.exceptions.ConnectionClosed` exceptions.
  
          .. _EAFP: https://docs.python.org/3/glossary.html#term-eafp
  
          """
          return self.state is State.OPEN and not self.transfer_data_task.done()
  
      @property
      def closed(self) -> bool:
          """
          :obj:`True` when the connection is closed; :obj:`False` otherwise.
  
          Be aware that both :attr:`open` and :attr:`closed` are :obj:`False`
          during the opening and closing sequences.
  
          """
          return self.state is State.CLOSED
  
      @property
      def close_code(self) -> int | None:
          """
          WebSocket close code, defined in `section 7.1.5 of RFC 6455`_.
  
          .. _section 7.1.5 of RFC 6455:
              https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.5
  
          :obj:`None` if the connection isn't closed yet.
  
          """
          if self.state is not State.CLOSED:
              return None
          elif self.close_rcvd is None:
              return CloseCode.ABNORMAL_CLOSURE
          else:
              return self.close_rcvd.code
  
      @property
      def close_reason(self) -> str | None:
          """
          WebSocket close reason, defined in `section 7.1.6 of RFC 6455`_.
  
          .. _section 7.1.6 of RFC 6455:
              https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.6
  
          :obj:`None` if the connection isn't closed yet.
  
          """
          if self.state is not State.CLOSED:
              return None
          elif self.close_rcvd is None:
              return ""
          else:
              return self.close_rcvd.reason
  
      async def __aiter__(self) -> AsyncIterator[Data]:
          """
          Iterate on incoming messages.
  
          The iterator exits normally when the connection is closed with the close
          code 1000 (OK) or 1001 (going away) or without a close code.
  
          It raises a :exc:`~websockets.exceptions.ConnectionClosedError`
          exception when the connection is closed with any other code.
  
          """
          try:
              while True:
                  yield await self.recv()
          except ConnectionClosedOK:
              return
  
      async def recv(self) -> Data:
          """
          Receive the next message.
  
          When the connection is closed, :meth:`recv` raises
          :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it raises
          :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal
          connection closure and
          :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol
          error or a network failure. This is how you detect the end of the
          message stream.
  
          Canceling :meth:`recv` is safe. There's no risk of losing the next
          message. The next invocation of :meth:`recv` will return it.
  
          This makes it possible to enforce a timeout by wrapping :meth:`recv` in
          :func:`~asyncio.timeout` or :func:`~asyncio.wait_for`.
  
          Returns:
              A string (:class:`str`) for a Text_ frame. A bytestring
              (:class:`bytes`) for a Binary_ frame.
  
              .. _Text: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
              .. _Binary: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  
          Raises:
              ConnectionClosed: When the connection is closed.
              RuntimeError: If two coroutines call :meth:`recv` concurrently.
  
          """
          if self._pop_message_waiter is not None:
              raise RuntimeError(
                  "cannot call recv while another coroutine "
                  "is already waiting for the next message"
              )
  
          # Don't await self.ensure_open() here:
          # - messages could be available in the queue even if the connection
          #   is closed;
          # - messages could be received before the closing frame even if the
          #   connection is closing.
  
          # Wait until there's a message in the queue (if necessary) or the
          # connection is closed.
          while len(self.messages) <= 0:
              pop_message_waiter: asyncio.Future[None] = self.loop.create_future()
              self._pop_message_waiter = pop_message_waiter
              try:
                  # If asyncio.wait() is canceled, it doesn't cancel
                  # pop_message_waiter and self.transfer_data_task.
                  await asyncio.wait(
                      [pop_message_waiter, self.transfer_data_task],
                      return_when=asyncio.FIRST_COMPLETED,
                  )
              finally:
                  self._pop_message_waiter = None
  
              # If asyncio.wait(...) exited because self.transfer_data_task
              # completed before receiving a new message, raise a suitable
              # exception (or return None if legacy_recv is enabled).
              if not pop_message_waiter.done():
                  if self.legacy_recv:
                      return None  # type: ignore
                  else:
                      # Wait until the connection is closed to raise
                      # ConnectionClosed with the correct code and reason.
                      await self.ensure_open()
  
          # Pop a message from the queue.
          message = self.messages.popleft()
  
          # Notify transfer_data().
          if self._put_message_waiter is not None:
              self._put_message_waiter.set_result(None)
              self._put_message_waiter = None
  
          return message
  
      async def send(
          self,
          message: DataLike | Iterable[DataLike] | AsyncIterable[DataLike],
      ) -> None:
          """
          Send a message.
  
          A string (:class:`str`) is sent as a Text_ frame. A bytestring or
          bytes-like object (:class:`bytes`, :class:`bytearray`, or
          :class:`memoryview`) is sent as a Binary_ frame.
  
          .. _Text: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
          .. _Binary: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  
          :meth:`send` also accepts an iterable or an asynchronous iterable of
          strings, bytestrings, or bytes-like objects to enable fragmentation_.
          Each item is treated as a message fragment and sent in its own frame.
          All items must be of the same type, or else :meth:`send` will raise a
          :exc:`TypeError` and the connection will be closed.
  
          .. _fragmentation: https://datatracker.ietf.org/doc/html/rfc6455#section-5.4
  
          :meth:`send` rejects dict-like objects because this is often an error.
          (If you want to send the keys of a dict-like object as fragments, call
          its :meth:`~dict.keys` method and pass the result to :meth:`send`.)
  
          Canceling :meth:`send` is discouraged. Instead, you should close the
          connection with :meth:`close`. Indeed, there are only two situations
          where :meth:`send` may yield control to the event loop and then get
          canceled; in both cases, :meth:`close` has the same effect and is
          more clear:
  
          1. The write buffer is full. If you don't want to wait until enough
             data is sent, your only alternative is to close the connection.
             :meth:`close` will likely time out then abort the TCP connection.
          2. ``message`` is an asynchronous iterator that yields control.
             Stopping in the middle of a fragmented message will cause a
             protocol error and the connection will be closed.
  
          When the connection is closed, :meth:`send` raises
          :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it
          raises :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal
          connection closure and
          :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol
          error or a network failure.
  
          Args:
              message: Message to send.
  
          Raises:
              ConnectionClosed: When the connection is closed.
              TypeError: If ``message`` doesn't have a supported type.
  
          """
          await self.ensure_open()
  
          # While sending a fragmented message, prevent sending other messages
          # until all fragments are sent.
          while self._fragmented_message_waiter is not None:
              await asyncio.shield(self._fragmented_message_waiter)
  
          # Unfragmented message -- this case must be handled first because
          # strings and bytes-like objects are iterable.
  
          if isinstance(message, (str, bytes, bytearray, memoryview)):
              opcode, data = prepare_data(message)
              await self.write_frame(True, opcode, data)
  
          # Catch a common mistake -- passing a dict to send().
  
          elif isinstance(message, Mapping):
              raise TypeError("data is a dict-like object")
  
          # Fragmented message -- regular iterator.
  
          elif isinstance(message, Iterable):
              iter_message = iter(message)
              try:
                  fragment = next(iter_message)
              except StopIteration:
                  return
              opcode, data = prepare_data(fragment)
  
              self._fragmented_message_waiter = self.loop.create_future()
              try:
                  # First fragment.
                  await self.write_frame(False, opcode, data)
  
                  # Other fragments.
                  for fragment in iter_message:
                      confirm_opcode, data = prepare_data(fragment)
                      if confirm_opcode != opcode:
                          raise TypeError("data contains inconsistent types")
                      await self.write_frame(False, OP_CONT, data)
  
                  # Final fragment.
                  await self.write_frame(True, OP_CONT, b"")
  
              except (Exception, asyncio.CancelledError):
                  # We're half-way through a fragmented message and we can't
                  # complete it. This makes the connection unusable.
                  self.fail_connection(CloseCode.INTERNAL_ERROR)
                  raise
  
              finally:
                  self._fragmented_message_waiter.set_result(None)
                  self._fragmented_message_waiter = None
  
          # Fragmented message -- asynchronous iterator
  
          elif isinstance(message, AsyncIterable):
              # Implement aiter_message = aiter(message) without aiter
              # Work around https://github.com/python/mypy/issues/5738
              aiter_message = cast(
                  Callable[[AsyncIterable[DataLike]], AsyncIterator[DataLike]],
                  type(message).__aiter__,
              )(message)
              try:
                  # Implement fragment = anext(aiter_message) without anext
                  # Work around https://github.com/python/mypy/issues/5738
                  fragment = await cast(
                      Callable[[AsyncIterator[DataLike]], Awaitable[DataLike]],
                      type(aiter_message).__anext__,
                  )(aiter_message)
              except StopAsyncIteration:
                  return
              opcode, data = prepare_data(fragment)
  
              self._fragmented_message_waiter = self.loop.create_future()
              try:
                  # First fragment.
                  await self.write_frame(False, opcode, data)
  
                  # Other fragments.
                  async for fragment in aiter_message:
                      confirm_opcode, data = prepare_data(fragment)
                      if confirm_opcode != opcode:
                          raise TypeError("data contains inconsistent types")
                      await self.write_frame(False, OP_CONT, data)
  
                  # Final fragment.
                  await self.write_frame(True, OP_CONT, b"")
  
              except (Exception, asyncio.CancelledError):
                  # We're half-way through a fragmented message and we can't
                  # complete it. This makes the connection unusable.
                  self.fail_connection(CloseCode.INTERNAL_ERROR)
                  raise
  
              finally:
                  self._fragmented_message_waiter.set_result(None)
                  self._fragmented_message_waiter = None
  
          else:
              raise TypeError("data must be str, bytes-like, or iterable")
  
      async def close(
          self,
          code: int = CloseCode.NORMAL_CLOSURE,
          reason: str = "",
      ) -> None:
          """
          Perform the closing handshake.
  
          :meth:`close` waits for the other end to complete the handshake and
          for the TCP connection to terminate. As a consequence, there's no need
          to await :meth:`wait_closed` after :meth:`close`.
  
          :meth:`close` is idempotent: it doesn't do anything once the
          connection is closed.
  
          Wrapping :func:`close` in :func:`~asyncio.create_task` is safe, given
          that errors during connection termination aren't particularly useful.
  
          Canceling :meth:`close` is discouraged. If it takes too long, you can
          set a shorter ``close_timeout``. If you don't want to wait, let the
          Python process exit, then the OS will take care of closing the TCP
          connection.
  
          Args:
              code: WebSocket close code.
              reason: WebSocket close reason.
  
          """
          try:
              async with asyncio_timeout(self.close_timeout):
                  await self.write_close_frame(Close(code, reason))
          except asyncio.TimeoutError:
              # If the close frame cannot be sent because the send buffers
              # are full, the closing handshake won't complete anyway.
              # Fail the connection to shut down faster.
              self.fail_connection()
  
          # If no close frame is received within the timeout, asyncio_timeout()
          # cancels the data transfer task and raises TimeoutError.
  
          # If close() is called multiple times concurrently and one of these
          # calls hits the timeout, the data transfer task will be canceled.
          # Other calls will receive a CancelledError here.
  
          try:
              # If close() is canceled during the wait, self.transfer_data_task
              # is canceled before the timeout elapses.
              async with asyncio_timeout(self.close_timeout):
                  await self.transfer_data_task
          except (asyncio.TimeoutError, asyncio.CancelledError):
              pass
  
          # Wait for the close connection task to close the TCP connection.
          await asyncio.shield(self.close_connection_task)
  
      async def wait_closed(self) -> None:
          """
          Wait until the connection is closed.
  
          This coroutine is identical to the :attr:`closed` attribute, except it
          can be awaited.
  
          This can make it easier to detect connection termination, regardless
          of its cause, in tasks that interact with the WebSocket connection.
  
          """
          await asyncio.shield(self.connection_lost_waiter)
  
      async def ping(self, data: DataLike | None = None) -> Awaitable[float]:
          """
          Send a Ping_.
  
          .. _Ping: https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.2
  
          A ping may serve as a keepalive, as a check that the remote endpoint
          received all messages up to this point, or to measure :attr:`latency`.
  
          Canceling :meth:`ping` is discouraged. If :meth:`ping` doesn't return
          immediately, it means the write buffer is full. If you don't want to
          wait, you should close the connection.
  
          Canceling the :class:`~asyncio.Future` returned by :meth:`ping` has no
          effect.
  
          Args:
              data: Payload of the ping. A string will be encoded to UTF-8.
                  If ``data`` is :obj:`None`, the payload is four random bytes.
  
          Returns:
              A future that will be completed when the corresponding pong is
              received. You can ignore it if you don't intend to wait. The result
              of the future is the latency of the connection in seconds.
  
              ::
  
                  pong_waiter = await ws.ping()
                  # only if you want to wait for the corresponding pong
                  latency = await pong_waiter
  
          Raises:
              ConnectionClosed: When the connection is closed.
              RuntimeError: If another ping was sent with the same data and
                  the corresponding pong wasn't received yet.
  
          """
          await self.ensure_open()
  
          if data is not None:
              data = prepare_ctrl(data)
  
          # Protect against duplicates if a payload is explicitly set.
          if data in self.pings:
  