protocol.py 62 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635
  1. from __future__ import annotations
  2. import asyncio
  3. import codecs
  4. import collections
  5. import logging
  6. import random
  7. import ssl
  8. import struct
  9. import sys
  10. import time
  11. import traceback
  12. import uuid
  13. import warnings
  14. from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Iterable, Mapping
  15. from typing import Any, Callable, Deque, cast
  16. from ..asyncio.compatibility import asyncio_timeout
  17. from ..datastructures import Headers
  18. from ..exceptions import (
  19. ConnectionClosed,
  20. ConnectionClosedError,
  21. ConnectionClosedOK,
  22. InvalidState,
  23. PayloadTooBig,
  24. ProtocolError,
  25. )
  26. from ..extensions import Extension
  27. from ..frames import (
  28. OK_CLOSE_CODES,
  29. OP_BINARY,
  30. OP_CLOSE,
  31. OP_CONT,
  32. OP_PING,
  33. OP_PONG,
  34. OP_TEXT,
  35. Close,
  36. CloseCode,
  37. Opcode,
  38. )
  39. from ..protocol import State
  40. from ..typing import BytesLike, Data, DataLike, LoggerLike, Subprotocol
  41. from .framing import Frame, prepare_ctrl, prepare_data
  42. __all__ = ["WebSocketCommonProtocol"]
  43. # In order to ensure consistency, the code always checks the current value of
  44. # WebSocketCommonProtocol.state before assigning a new value and never yields
  45. # between the check and the assignment.
  46. class WebSocketCommonProtocol(asyncio.Protocol):
  47. """
  48. WebSocket connection.
  49. :class:`WebSocketCommonProtocol` provides APIs shared between WebSocket
  50. servers and clients. You shouldn't use it directly. Instead, use
  51. :class:`~websockets.legacy.client.WebSocketClientProtocol` or
  52. :class:`~websockets.legacy.server.WebSocketServerProtocol`.
  53. This documentation focuses on low-level details that aren't covered in the
  54. documentation of :class:`~websockets.legacy.client.WebSocketClientProtocol`
  55. and :class:`~websockets.legacy.server.WebSocketServerProtocol` for the sake
  56. of simplicity.
  57. Once the connection is open, a Ping_ frame is sent every ``ping_interval``
  58. seconds. This serves as a keepalive. It helps keeping the connection open,
  59. especially in the presence of proxies with short timeouts on inactive
  60. connections. Set ``ping_interval`` to :obj:`None` to disable this behavior.
  61. .. _Ping: https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.2
  62. If the corresponding Pong_ frame isn't received within ``ping_timeout``
  63. seconds, the connection is considered unusable and is closed with code 1011.
  64. This ensures that the remote endpoint remains responsive. Set
  65. ``ping_timeout`` to :obj:`None` to disable this behavior.
  66. .. _Pong: https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.3
  67. See the discussion of :doc:`keepalive <../../topics/keepalive>` for details.
  68. The ``close_timeout`` parameter defines a maximum wait time for completing
  69. the closing handshake and terminating the TCP connection. For legacy
  70. reasons, :meth:`close` completes in at most ``5 * close_timeout`` seconds
  71. for clients and ``4 * close_timeout`` for servers.
  72. ``close_timeout`` is a parameter of the protocol because websockets usually
  73. calls :meth:`close` implicitly upon exit:
  74. * on the client side, when using :func:`~websockets.legacy.client.connect`
  75. as a context manager;
  76. * on the server side, when the connection handler terminates.
  77. To apply a timeout to any other API, wrap it in :func:`~asyncio.timeout` or
  78. :func:`~asyncio.wait_for`.
  79. The ``max_size`` parameter enforces the maximum size for incoming messages
  80. in bytes. The default value is 1 MiB. If a larger message is received,
  81. :meth:`recv` will raise :exc:`~websockets.exceptions.ConnectionClosedError`
  82. and the connection will be closed with code 1009.
  83. The ``max_queue`` parameter sets the maximum length of the queue that
  84. holds incoming messages. The default value is ``32``. Messages are added
  85. to an in-memory queue when they're received; then :meth:`recv` pops from
  86. that queue. In order to prevent excessive memory consumption when
  87. messages are received faster than they can be processed, the queue must
  88. be bounded. If the queue fills up, the protocol stops processing incoming
  89. data until :meth:`recv` is called. In this situation, various receive
  90. buffers (at least in :mod:`asyncio` and in the OS) will fill up, then the
  91. TCP receive window will shrink, slowing down transmission to avoid packet
  92. loss.
  93. Since Python can use up to 4 bytes of memory to represent a single
  94. character, each connection may use up to ``4 * max_size * max_queue``
  95. bytes of memory to store incoming messages. By default, this is 128 MiB.
  96. You may want to lower the limits, depending on your application's
  97. requirements.
  98. The ``read_limit`` argument sets the high-water limit of the buffer for
  99. incoming bytes. The low-water limit is half the high-water limit. The
  100. default value is 64 KiB, half of asyncio's default (based on the current
  101. implementation of :class:`~asyncio.StreamReader`).
  102. The ``write_limit`` argument sets the high-water limit of the buffer for
  103. outgoing bytes. The low-water limit is a quarter of the high-water limit.
  104. The default value is 64 KiB, equal to asyncio's default (based on the
  105. current implementation of ``FlowControlMixin``).
  106. See the discussion of :doc:`memory usage <../../topics/memory>` for details.
  107. Args:
  108. logger: Logger for this server.
  109. It defaults to ``logging.getLogger("websockets.protocol")``.
  110. See the :doc:`logging guide <../../topics/logging>` for details.
  111. ping_interval: Interval between keepalive pings in seconds.
  112. :obj:`None` disables keepalive.
  113. ping_timeout: Timeout for keepalive pings in seconds.
  114. :obj:`None` disables timeouts.
  115. close_timeout: Timeout for closing the connection in seconds.
  116. For legacy reasons, the actual timeout is 4 or 5 times larger.
  117. max_size: Maximum size of incoming messages in bytes.
  118. :obj:`None` disables the limit.
  119. max_queue: Maximum number of incoming messages in receive buffer.
  120. :obj:`None` disables the limit.
  121. read_limit: High-water mark of read buffer in bytes.
  122. write_limit: High-water mark of write buffer in bytes.
  123. """
  124. # There are only two differences between the client-side and server-side
  125. # behavior: masking the payload and closing the underlying TCP connection.
  126. # Set is_client = True/False and side = "client"/"server" to pick a side.
  127. is_client: bool
  128. side: str = "undefined"
  129. def __init__(
  130. self,
  131. *,
  132. logger: LoggerLike | None = None,
  133. ping_interval: float | None = 20,
  134. ping_timeout: float | None = 20,
  135. close_timeout: float | None = None,
  136. max_size: int | None = 2**20,
  137. max_queue: int | None = 2**5,
  138. read_limit: int = 2**16,
  139. write_limit: int = 2**16,
  140. # The following arguments are kept only for backwards compatibility.
  141. host: str | None = None,
  142. port: int | None = None,
  143. secure: bool | None = None,
  144. legacy_recv: bool = False,
  145. loop: asyncio.AbstractEventLoop | None = None,
  146. timeout: float | None = None,
  147. ) -> None:
  148. if legacy_recv: # pragma: no cover
  149. warnings.warn("legacy_recv is deprecated", DeprecationWarning)
  150. # Backwards compatibility: close_timeout used to be called timeout.
  151. if timeout is None:
  152. timeout = 10
  153. else:
  154. warnings.warn("rename timeout to close_timeout", DeprecationWarning)
  155. # If both are specified, timeout is ignored.
  156. if close_timeout is None:
  157. close_timeout = timeout
  158. # Backwards compatibility: the loop parameter used to be supported.
  159. if loop is None:
  160. loop = asyncio.get_event_loop()
  161. else:
  162. warnings.warn("remove loop argument", DeprecationWarning)
  163. self.ping_interval = ping_interval
  164. self.ping_timeout = ping_timeout
  165. self.close_timeout = close_timeout
  166. self.max_size = max_size
  167. self.max_queue = max_queue
  168. self.read_limit = read_limit
  169. self.write_limit = write_limit
  170. # Unique identifier. For logs.
  171. self.id: uuid.UUID = uuid.uuid4()
  172. """Unique identifier of the connection. Useful in logs."""
  173. # Logger or LoggerAdapter for this connection.
  174. if logger is None:
  175. logger = logging.getLogger("websockets.protocol")
  176. self.logger: LoggerLike = logging.LoggerAdapter(logger, {"websocket": self})
  177. """Logger for this connection."""
  178. # Track if DEBUG is enabled. Shortcut logging calls if it isn't.
  179. self.debug = logger.isEnabledFor(logging.DEBUG)
  180. self.loop = loop
  181. self._host = host
  182. self._port = port
  183. self._secure = secure
  184. self.legacy_recv = legacy_recv
  185. # Configure read buffer limits. The high-water limit is defined by
  186. # ``self.read_limit``. The ``limit`` argument controls the line length
  187. # limit and half the buffer limit of :class:`~asyncio.StreamReader`.
  188. # That's why it must be set to half of ``self.read_limit``.
  189. self.reader = asyncio.StreamReader(limit=read_limit // 2, loop=loop)
  190. # Copied from asyncio.FlowControlMixin
  191. self._paused = False
  192. self._drain_waiter: asyncio.Future[None] | None = None
  193. # This class implements the data transfer and closing handshake, which
  194. # are shared between the client-side and the server-side.
  195. # Subclasses implement the opening handshake and, on success, execute
  196. # :meth:`connection_open` to change the state to OPEN.
  197. self.state = State.CONNECTING
  198. if self.debug:
  199. self.logger.debug("= connection is CONNECTING")
  200. # HTTP protocol parameters.
  201. self.path: str
  202. """Path of the opening handshake request."""
  203. self.request_headers: Headers
  204. """Opening handshake request headers."""
  205. self.response_headers: Headers
  206. """Opening handshake response headers."""
  207. # WebSocket protocol parameters.
  208. self.extensions: list[Extension] = []
  209. self.subprotocol: Subprotocol | None = None
  210. """Subprotocol, if one was negotiated."""
  211. # Close code and reason, set when a close frame is sent or received.
  212. self.close_rcvd: Close | None = None
  213. self.close_sent: Close | None = None
  214. self.close_rcvd_then_sent: bool | None = None
  215. # Completed when the connection state becomes CLOSED. Translates the
  216. # :meth:`connection_lost` callback to a :class:`~asyncio.Future`
  217. # that can be awaited. (Other :class:`~asyncio.Protocol` callbacks are
  218. # translated by ``self.stream_reader``).
  219. self.connection_lost_waiter: asyncio.Future[None] = loop.create_future()
  220. # Queue of received messages.
  221. self.messages: Deque[Data] = collections.deque()
  222. self._pop_message_waiter: asyncio.Future[None] | None = None
  223. self._put_message_waiter: asyncio.Future[None] | None = None
  224. # Protect sending fragmented messages.
  225. self._fragmented_message_waiter: asyncio.Future[None] | None = None
  226. # Mapping of ping IDs to pong waiters, in chronological order.
  227. self.pings: dict[bytes, tuple[asyncio.Future[float], float]] = {}
  228. self.latency: float = 0
  229. """
  230. Latency of the connection, in seconds.
  231. Latency is defined as the round-trip time of the connection. It is
  232. measured by sending a Ping frame and waiting for a matching Pong frame.
  233. Before the first measurement, :attr:`latency` is ``0``.
  234. By default, websockets enables a :ref:`keepalive <keepalive>` mechanism
  235. that sends Ping frames automatically at regular intervals. You can also
  236. send Ping frames and measure latency with :meth:`ping`.
  237. """
  238. # Task running the data transfer.
  239. self.transfer_data_task: asyncio.Task[None]
  240. # Exception that occurred during data transfer, if any.
  241. self.transfer_data_exc: BaseException | None = None
  242. # Task sending keepalive pings.
  243. self.keepalive_ping_task: asyncio.Task[None]
  244. # Task closing the TCP connection.
  245. self.close_connection_task: asyncio.Task[None]
  246. # Copied from asyncio.FlowControlMixin
  247. async def _drain_helper(self) -> None: # pragma: no cover
  248. if self.connection_lost_waiter.done():
  249. raise ConnectionResetError("Connection lost")
  250. if not self._paused:
  251. return
  252. waiter = self._drain_waiter
  253. assert waiter is None or waiter.cancelled()
  254. waiter = self.loop.create_future()
  255. self._drain_waiter = waiter
  256. await waiter
  257. # Copied from asyncio.StreamWriter
  258. async def _drain(self) -> None: # pragma: no cover
  259. if self.reader is not None:
  260. exc = self.reader.exception()
  261. if exc is not None:
  262. raise exc
  263. if self.transport is not None:
  264. if self.transport.is_closing():
  265. # Yield to the event loop so connection_lost() may be
  266. # called. Without this, _drain_helper() would return
  267. # immediately, and code that calls
  268. # write(...); yield from drain()
  269. # in a loop would never call connection_lost(), so it
  270. # would not see an error when the socket is closed.
  271. await asyncio.sleep(0)
  272. await self._drain_helper()
  273. def connection_open(self) -> None:
  274. """
  275. Callback when the WebSocket opening handshake completes.
  276. Enter the OPEN state and start the data transfer phase.
  277. """
  278. # 4.1. The WebSocket Connection is Established.
  279. assert self.state is State.CONNECTING
  280. self.state = State.OPEN
  281. if self.debug:
  282. self.logger.debug("= connection is OPEN")
  283. # Start the task that receives incoming WebSocket messages.
  284. self.transfer_data_task = self.loop.create_task(self.transfer_data())
  285. # Start the task that sends pings at regular intervals.
  286. self.keepalive_ping_task = self.loop.create_task(self.keepalive_ping())
  287. # Start the task that eventually closes the TCP connection.
  288. self.close_connection_task = self.loop.create_task(self.close_connection())
  289. @property
  290. def host(self) -> str | None:
  291. alternative = "remote_address" if self.is_client else "local_address"
  292. warnings.warn(f"use {alternative}[0] instead of host", DeprecationWarning)
  293. return self._host
  294. @property
  295. def port(self) -> int | None:
  296. alternative = "remote_address" if self.is_client else "local_address"
  297. warnings.warn(f"use {alternative}[1] instead of port", DeprecationWarning)
  298. return self._port
  299. @property
  300. def secure(self) -> bool | None:
  301. warnings.warn("don't use secure", DeprecationWarning)
  302. return self._secure
  303. # Public API
  304. @property
  305. def local_address(self) -> Any:
  306. """
  307. Local address of the connection.
  308. For IPv4 connections, this is a ``(host, port)`` tuple.
  309. The format of the address depends on the address family;
  310. see :meth:`~socket.socket.getsockname`.
  311. :obj:`None` if the TCP connection isn't established yet.
  312. """
  313. try:
  314. transport = self.transport
  315. except AttributeError:
  316. return None
  317. else:
  318. return transport.get_extra_info("sockname")
  319. @property
  320. def remote_address(self) -> Any:
  321. """
  322. Remote address of the connection.
  323. For IPv4 connections, this is a ``(host, port)`` tuple.
  324. The format of the address depends on the address family;
  325. see :meth:`~socket.socket.getpeername`.
  326. :obj:`None` if the TCP connection isn't established yet.
  327. """
  328. try:
  329. transport = self.transport
  330. except AttributeError:
  331. return None
  332. else:
  333. return transport.get_extra_info("peername")
  334. @property
  335. def open(self) -> bool:
  336. """
  337. :obj:`True` when the connection is open; :obj:`False` otherwise.
  338. This attribute may be used to detect disconnections. However, this
  339. approach is discouraged per the EAFP_ principle. Instead, you should
  340. handle :exc:`~websockets.exceptions.ConnectionClosed` exceptions.
  341. .. _EAFP: https://docs.python.org/3/glossary.html#term-eafp
  342. """
  343. return self.state is State.OPEN and not self.transfer_data_task.done()
  344. @property
  345. def closed(self) -> bool:
  346. """
  347. :obj:`True` when the connection is closed; :obj:`False` otherwise.
  348. Be aware that both :attr:`open` and :attr:`closed` are :obj:`False`
  349. during the opening and closing sequences.
  350. """
  351. return self.state is State.CLOSED
  352. @property
  353. def close_code(self) -> int | None:
  354. """
  355. WebSocket close code, defined in `section 7.1.5 of RFC 6455`_.
  356. .. _section 7.1.5 of RFC 6455:
  357. https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.5
  358. :obj:`None` if the connection isn't closed yet.
  359. """
  360. if self.state is not State.CLOSED:
  361. return None
  362. elif self.close_rcvd is None:
  363. return CloseCode.ABNORMAL_CLOSURE
  364. else:
  365. return self.close_rcvd.code
  366. @property
  367. def close_reason(self) -> str | None:
  368. """
  369. WebSocket close reason, defined in `section 7.1.6 of RFC 6455`_.
  370. .. _section 7.1.6 of RFC 6455:
  371. https://datatracker.ietf.org/doc/html/rfc6455#section-7.1.6
  372. :obj:`None` if the connection isn't closed yet.
  373. """
  374. if self.state is not State.CLOSED:
  375. return None
  376. elif self.close_rcvd is None:
  377. return ""
  378. else:
  379. return self.close_rcvd.reason
  380. async def __aiter__(self) -> AsyncIterator[Data]:
  381. """
  382. Iterate on incoming messages.
  383. The iterator exits normally when the connection is closed with the close
  384. code 1000 (OK) or 1001 (going away) or without a close code.
  385. It raises a :exc:`~websockets.exceptions.ConnectionClosedError`
  386. exception when the connection is closed with any other code.
  387. """
  388. try:
  389. while True:
  390. yield await self.recv()
  391. except ConnectionClosedOK:
  392. return
  393. async def recv(self) -> Data:
  394. """
  395. Receive the next message.
  396. When the connection is closed, :meth:`recv` raises
  397. :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it raises
  398. :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal
  399. connection closure and
  400. :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol
  401. error or a network failure. This is how you detect the end of the
  402. message stream.
  403. Canceling :meth:`recv` is safe. There's no risk of losing the next
  404. message. The next invocation of :meth:`recv` will return it.
  405. This makes it possible to enforce a timeout by wrapping :meth:`recv` in
  406. :func:`~asyncio.timeout` or :func:`~asyncio.wait_for`.
  407. Returns:
  408. A string (:class:`str`) for a Text_ frame. A bytestring
  409. (:class:`bytes`) for a Binary_ frame.
  410. .. _Text: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  411. .. _Binary: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  412. Raises:
  413. ConnectionClosed: When the connection is closed.
  414. RuntimeError: If two coroutines call :meth:`recv` concurrently.
  415. """
  416. if self._pop_message_waiter is not None:
  417. raise RuntimeError(
  418. "cannot call recv while another coroutine "
  419. "is already waiting for the next message"
  420. )
  421. # Don't await self.ensure_open() here:
  422. # - messages could be available in the queue even if the connection
  423. # is closed;
  424. # - messages could be received before the closing frame even if the
  425. # connection is closing.
  426. # Wait until there's a message in the queue (if necessary) or the
  427. # connection is closed.
  428. while len(self.messages) <= 0:
  429. pop_message_waiter: asyncio.Future[None] = self.loop.create_future()
  430. self._pop_message_waiter = pop_message_waiter
  431. try:
  432. # If asyncio.wait() is canceled, it doesn't cancel
  433. # pop_message_waiter and self.transfer_data_task.
  434. await asyncio.wait(
  435. [pop_message_waiter, self.transfer_data_task],
  436. return_when=asyncio.FIRST_COMPLETED,
  437. )
  438. finally:
  439. self._pop_message_waiter = None
  440. # If asyncio.wait(...) exited because self.transfer_data_task
  441. # completed before receiving a new message, raise a suitable
  442. # exception (or return None if legacy_recv is enabled).
  443. if not pop_message_waiter.done():
  444. if self.legacy_recv:
  445. return None # type: ignore
  446. else:
  447. # Wait until the connection is closed to raise
  448. # ConnectionClosed with the correct code and reason.
  449. await self.ensure_open()
  450. # Pop a message from the queue.
  451. message = self.messages.popleft()
  452. # Notify transfer_data().
  453. if self._put_message_waiter is not None:
  454. self._put_message_waiter.set_result(None)
  455. self._put_message_waiter = None
  456. return message
  457. async def send(
  458. self,
  459. message: DataLike | Iterable[DataLike] | AsyncIterable[DataLike],
  460. ) -> None:
  461. """
  462. Send a message.
  463. A string (:class:`str`) is sent as a Text_ frame. A bytestring or
  464. bytes-like object (:class:`bytes`, :class:`bytearray`, or
  465. :class:`memoryview`) is sent as a Binary_ frame.
  466. .. _Text: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  467. .. _Binary: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  468. :meth:`send` also accepts an iterable or an asynchronous iterable of
  469. strings, bytestrings, or bytes-like objects to enable fragmentation_.
  470. Each item is treated as a message fragment and sent in its own frame.
  471. All items must be of the same type, or else :meth:`send` will raise a
  472. :exc:`TypeError` and the connection will be closed.
  473. .. _fragmentation: https://datatracker.ietf.org/doc/html/rfc6455#section-5.4
  474. :meth:`send` rejects dict-like objects because this is often an error.
  475. (If you want to send the keys of a dict-like object as fragments, call
  476. its :meth:`~dict.keys` method and pass the result to :meth:`send`.)
  477. Canceling :meth:`send` is discouraged. Instead, you should close the
  478. connection with :meth:`close`. Indeed, there are only two situations
  479. where :meth:`send` may yield control to the event loop and then get
  480. canceled; in both cases, :meth:`close` has the same effect and is
  481. more clear:
  482. 1. The write buffer is full. If you don't want to wait until enough
  483. data is sent, your only alternative is to close the connection.
  484. :meth:`close` will likely time out then abort the TCP connection.
  485. 2. ``message`` is an asynchronous iterator that yields control.
  486. Stopping in the middle of a fragmented message will cause a
  487. protocol error and the connection will be closed.
  488. When the connection is closed, :meth:`send` raises
  489. :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it
  490. raises :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal
  491. connection closure and
  492. :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol
  493. error or a network failure.
  494. Args:
  495. message: Message to send.
  496. Raises:
  497. ConnectionClosed: When the connection is closed.
  498. TypeError: If ``message`` doesn't have a supported type.
  499. """
  500. await self.ensure_open()
  501. # While sending a fragmented message, prevent sending other messages
  502. # until all fragments are sent.
  503. while self._fragmented_message_waiter is not None:
  504. await asyncio.shield(self._fragmented_message_waiter)
  505. # Unfragmented message -- this case must be handled first because
  506. # strings and bytes-like objects are iterable.
  507. if isinstance(message, (str, bytes, bytearray, memoryview)):
  508. opcode, data = prepare_data(message)
  509. await self.write_frame(True, opcode, data)
  510. # Catch a common mistake -- passing a dict to send().
  511. elif isinstance(message, Mapping):
  512. raise TypeError("data is a dict-like object")
  513. # Fragmented message -- regular iterator.
  514. elif isinstance(message, Iterable):
  515. iter_message = iter(message)
  516. try:
  517. fragment = next(iter_message)
  518. except StopIteration:
  519. return
  520. opcode, data = prepare_data(fragment)
  521. self._fragmented_message_waiter = self.loop.create_future()
  522. try:
  523. # First fragment.
  524. await self.write_frame(False, opcode, data)
  525. # Other fragments.
  526. for fragment in iter_message:
  527. confirm_opcode, data = prepare_data(fragment)
  528. if confirm_opcode != opcode:
  529. raise TypeError("data contains inconsistent types")
  530. await self.write_frame(False, OP_CONT, data)
  531. # Final fragment.
  532. await self.write_frame(True, OP_CONT, b"")
  533. except (Exception, asyncio.CancelledError):
  534. # We're half-way through a fragmented message and we can't
  535. # complete it. This makes the connection unusable.
  536. self.fail_connection(CloseCode.INTERNAL_ERROR)
  537. raise
  538. finally:
  539. self._fragmented_message_waiter.set_result(None)
  540. self._fragmented_message_waiter = None
  541. # Fragmented message -- asynchronous iterator
  542. elif isinstance(message, AsyncIterable):
  543. # Implement aiter_message = aiter(message) without aiter
  544. # Work around https://github.com/python/mypy/issues/5738
  545. aiter_message = cast(
  546. Callable[[AsyncIterable[DataLike]], AsyncIterator[DataLike]],
  547. type(message).__aiter__,
  548. )(message)
  549. try:
  550. # Implement fragment = anext(aiter_message) without anext
  551. # Work around https://github.com/python/mypy/issues/5738
  552. fragment = await cast(
  553. Callable[[AsyncIterator[DataLike]], Awaitable[DataLike]],
  554. type(aiter_message).__anext__,
  555. )(aiter_message)
  556. except StopAsyncIteration:
  557. return
  558. opcode, data = prepare_data(fragment)
  559. self._fragmented_message_waiter = self.loop.create_future()
  560. try:
  561. # First fragment.
  562. await self.write_frame(False, opcode, data)
  563. # Other fragments.
  564. async for fragment in aiter_message:
  565. confirm_opcode, data = prepare_data(fragment)
  566. if confirm_opcode != opcode:
  567. raise TypeError("data contains inconsistent types")
  568. await self.write_frame(False, OP_CONT, data)
  569. # Final fragment.
  570. await self.write_frame(True, OP_CONT, b"")
  571. except (Exception, asyncio.CancelledError):
  572. # We're half-way through a fragmented message and we can't
  573. # complete it. This makes the connection unusable.
  574. self.fail_connection(CloseCode.INTERNAL_ERROR)
  575. raise
  576. finally:
  577. self._fragmented_message_waiter.set_result(None)
  578. self._fragmented_message_waiter = None
  579. else:
  580. raise TypeError("data must be str, bytes-like, or iterable")
  581. async def close(
  582. self,
  583. code: int = CloseCode.NORMAL_CLOSURE,
  584. reason: str = "",
  585. ) -> None:
  586. """
  587. Perform the closing handshake.
  588. :meth:`close` waits for the other end to complete the handshake and
  589. for the TCP connection to terminate. As a consequence, there's no need
  590. to await :meth:`wait_closed` after :meth:`close`.
  591. :meth:`close` is idempotent: it doesn't do anything once the
  592. connection is closed.
  593. Wrapping :func:`close` in :func:`~asyncio.create_task` is safe, given
  594. that errors during connection termination aren't particularly useful.
  595. Canceling :meth:`close` is discouraged. If it takes too long, you can
  596. set a shorter ``close_timeout``. If you don't want to wait, let the
  597. Python process exit, then the OS will take care of closing the TCP
  598. connection.
  599. Args:
  600. code: WebSocket close code.
  601. reason: WebSocket close reason.
  602. """
  603. try:
  604. async with asyncio_timeout(self.close_timeout):
  605. await self.write_close_frame(Close(code, reason))
  606. except asyncio.TimeoutError:
  607. # If the close frame cannot be sent because the send buffers
  608. # are full, the closing handshake won't complete anyway.
  609. # Fail the connection to shut down faster.
  610. self.fail_connection()
  611. # If no close frame is received within the timeout, asyncio_timeout()
  612. # cancels the data transfer task and raises TimeoutError.
  613. # If close() is called multiple times concurrently and one of these
  614. # calls hits the timeout, the data transfer task will be canceled.
  615. # Other calls will receive a CancelledError here.
  616. try:
  617. # If close() is canceled during the wait, self.transfer_data_task
  618. # is canceled before the timeout elapses.
  619. async with asyncio_timeout(self.close_timeout):
  620. await self.transfer_data_task
  621. except (asyncio.TimeoutError, asyncio.CancelledError):
  622. pass
  623. # Wait for the close connection task to close the TCP connection.
  624. await asyncio.shield(self.close_connection_task)
  625. async def wait_closed(self) -> None:
  626. """
  627. Wait until the connection is closed.
  628. This coroutine is identical to the :attr:`closed` attribute, except it
  629. can be awaited.
  630. This can make it easier to detect connection termination, regardless
  631. of its cause, in tasks that interact with the WebSocket connection.
  632. """
  633. await asyncio.shield(self.connection_lost_waiter)
  634. async def ping(self, data: DataLike | None = None) -> Awaitable[float]:
  635. """
  636. Send a Ping_.
  637. .. _Ping: https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.2
  638. A ping may serve as a keepalive, as a check that the remote endpoint
  639. received all messages up to this point, or to measure :attr:`latency`.
  640. Canceling :meth:`ping` is discouraged. If :meth:`ping` doesn't return
  641. immediately, it means the write buffer is full. If you don't want to
  642. wait, you should close the connection.
  643. Canceling the :class:`~asyncio.Future` returned by :meth:`ping` has no
  644. effect.
  645. Args:
  646. data: Payload of the ping. A string will be encoded to UTF-8.
  647. If ``data`` is :obj:`None`, the payload is four random bytes.
  648. Returns:
  649. A future that will be completed when the corresponding pong is
  650. received. You can ignore it if you don't intend to wait. The result
  651. of the future is the latency of the connection in seconds.
  652. ::
  653. pong_waiter = await ws.ping()
  654. # only if you want to wait for the corresponding pong
  655. latency = await pong_waiter
  656. Raises:
  657. ConnectionClosed: When the connection is closed.
  658. RuntimeError: If another ping was sent with the same data and
  659. the corresponding pong wasn't received yet.
  660. """
  661. await self.ensure_open()
  662. if data is not None:
  663. data = prepare_ctrl(data)
  664. # Protect against duplicates if a payload is explicitly set.
  665. if data in self.pings:
  666. raise RuntimeError("already waiting for a pong with the same data")
  667. # Generate a unique random payload otherwise.
  668. while data is None or data in self.pings:
  669. data = struct.pack("!I", random.getrandbits(32))
  670. pong_waiter = self.loop.create_future()
  671. # Resolution of time.monotonic() may be too low on Windows.
  672. ping_timestamp = time.perf_counter()
  673. self.pings[data] = (pong_waiter, ping_timestamp)
  674. await self.write_frame(True, OP_PING, data)
  675. return asyncio.shield(pong_waiter)
  676. async def pong(self, data: DataLike = b"") -> None:
  677. """
  678. Send a Pong_.
  679. .. _Pong: https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.3
  680. An unsolicited pong may serve as a unidirectional heartbeat.
  681. Canceling :meth:`pong` is discouraged. If :meth:`pong` doesn't return
  682. immediately, it means the write buffer is full. If you don't want to
  683. wait, you should close the connection.
  684. Args:
  685. data: Payload of the pong. A string will be encoded to UTF-8.
  686. Raises:
  687. ConnectionClosed: When the connection is closed.
  688. """
  689. await self.ensure_open()
  690. data = prepare_ctrl(data)
  691. await self.write_frame(True, OP_PONG, data)
  692. # Private methods - no guarantees.
  693. def connection_closed_exc(self) -> ConnectionClosed:
  694. exc: ConnectionClosed
  695. if (
  696. self.close_rcvd is not None
  697. and self.close_rcvd.code in OK_CLOSE_CODES
  698. and self.close_sent is not None
  699. and self.close_sent.code in OK_CLOSE_CODES
  700. ):
  701. exc = ConnectionClosedOK(
  702. self.close_rcvd,
  703. self.close_sent,
  704. self.close_rcvd_then_sent,
  705. )
  706. else:
  707. exc = ConnectionClosedError(
  708. self.close_rcvd,
  709. self.close_sent,
  710. self.close_rcvd_then_sent,
  711. )
  712. # Chain to the exception that terminated data transfer, if any.
  713. exc.__cause__ = self.transfer_data_exc
  714. return exc
  715. async def ensure_open(self) -> None:
  716. """
  717. Check that the WebSocket connection is open.
  718. Raise :exc:`~websockets.exceptions.ConnectionClosed` if it isn't.
  719. """
  720. # Handle cases from most common to least common for performance.
  721. if self.state is State.OPEN:
  722. # If self.transfer_data_task exited without a closing handshake,
  723. # self.close_connection_task may be closing the connection, going
  724. # straight from OPEN to CLOSED.
  725. if self.transfer_data_task.done():
  726. await asyncio.shield(self.close_connection_task)
  727. raise self.connection_closed_exc()
  728. else:
  729. return
  730. if self.state is State.CLOSED:
  731. raise self.connection_closed_exc()
  732. if self.state is State.CLOSING:
  733. # If we started the closing handshake, wait for its completion to
  734. # get the proper close code and reason. self.close_connection_task
  735. # will complete within 4 or 5 * close_timeout after close(). The
  736. # CLOSING state also occurs when failing the connection. In that
  737. # case self.close_connection_task will complete even faster.
  738. await asyncio.shield(self.close_connection_task)
  739. raise self.connection_closed_exc()
  740. # Control may only reach this point in buggy third-party subclasses.
  741. assert self.state is State.CONNECTING
  742. raise InvalidState("WebSocket connection isn't established yet")
  743. async def transfer_data(self) -> None:
  744. """
  745. Read incoming messages and put them in a queue.
  746. This coroutine runs in a task until the closing handshake is started.
  747. """
  748. try:
  749. while True:
  750. message = await self.read_message()
  751. # Exit the loop when receiving a close frame.
  752. if message is None:
  753. break
  754. # Wait until there's room in the queue (if necessary).
  755. if self.max_queue is not None:
  756. while len(self.messages) >= self.max_queue:
  757. self._put_message_waiter = self.loop.create_future()
  758. try:
  759. await asyncio.shield(self._put_message_waiter)
  760. finally:
  761. self._put_message_waiter = None
  762. # Put the message in the queue.
  763. self.messages.append(message)
  764. # Notify recv().
  765. if self._pop_message_waiter is not None:
  766. self._pop_message_waiter.set_result(None)
  767. self._pop_message_waiter = None
  768. except asyncio.CancelledError as exc:
  769. self.transfer_data_exc = exc
  770. # If fail_connection() cancels this task, avoid logging the error
  771. # twice and failing the connection again.
  772. raise
  773. except ProtocolError as exc:
  774. self.transfer_data_exc = exc
  775. self.fail_connection(CloseCode.PROTOCOL_ERROR)
  776. except (ConnectionError, TimeoutError, EOFError, ssl.SSLError) as exc:
  777. # Reading data with self.reader.readexactly may raise:
  778. # - most subclasses of ConnectionError if the TCP connection
  779. # breaks, is reset, or is aborted;
  780. # - TimeoutError if the TCP connection times out;
  781. # - IncompleteReadError, a subclass of EOFError, if fewer
  782. # bytes are available than requested;
  783. # - ssl.SSLError if the other side infringes the TLS protocol.
  784. self.transfer_data_exc = exc
  785. self.fail_connection(CloseCode.ABNORMAL_CLOSURE)
  786. except UnicodeDecodeError as exc:
  787. self.transfer_data_exc = exc
  788. self.fail_connection(CloseCode.INVALID_DATA)
  789. except PayloadTooBig as exc:
  790. self.transfer_data_exc = exc
  791. self.fail_connection(CloseCode.MESSAGE_TOO_BIG)
  792. except Exception as exc:
  793. # This shouldn't happen often because exceptions expected under
  794. # regular circumstances are handled above. If it does, consider
  795. # catching and handling more exceptions.
  796. self.logger.error("data transfer failed", exc_info=True)
  797. self.transfer_data_exc = exc
  798. self.fail_connection(CloseCode.INTERNAL_ERROR)
  799. async def read_message(self) -> Data | None:
  800. """
  801. Read a single message from the connection.
  802. Re-assemble data frames if the message is fragmented.
  803. Return :obj:`None` when the closing handshake is started.
  804. """
  805. frame = await self.read_data_frame(max_size=self.max_size)
  806. # A close frame was received.
  807. if frame is None:
  808. return None
  809. if frame.opcode == OP_TEXT:
  810. text = True
  811. elif frame.opcode == OP_BINARY:
  812. text = False
  813. else: # frame.opcode == OP_CONT
  814. raise ProtocolError("unexpected opcode")
  815. # Shortcut for the common case - no fragmentation
  816. if frame.fin:
  817. if isinstance(frame.data, memoryview):
  818. raise AssertionError("only compressed outgoing frames use memoryview")
  819. return frame.data.decode() if text else bytes(frame.data)
  820. # 5.4. Fragmentation
  821. fragments: list[DataLike] = []
  822. max_size = self.max_size
  823. if text:
  824. decoder_factory = codecs.getincrementaldecoder("utf-8")
  825. decoder = decoder_factory(errors="strict")
  826. if max_size is None:
  827. def append(frame: Frame) -> None:
  828. nonlocal fragments
  829. fragments.append(decoder.decode(frame.data, frame.fin))
  830. else:
  831. def append(frame: Frame) -> None:
  832. nonlocal fragments, max_size
  833. fragments.append(decoder.decode(frame.data, frame.fin))
  834. assert isinstance(max_size, int)
  835. max_size -= len(frame.data)
  836. else:
  837. if max_size is None:
  838. def append(frame: Frame) -> None:
  839. nonlocal fragments
  840. fragments.append(frame.data)
  841. else:
  842. def append(frame: Frame) -> None:
  843. nonlocal fragments, max_size
  844. fragments.append(frame.data)
  845. assert isinstance(max_size, int)
  846. max_size -= len(frame.data)
  847. append(frame)
  848. while not frame.fin:
  849. frame = await self.read_data_frame(max_size=max_size)
  850. if frame is None:
  851. raise ProtocolError("incomplete fragmented message")
  852. if frame.opcode != OP_CONT:
  853. raise ProtocolError("unexpected opcode")
  854. append(frame)
  855. return ("" if text else b"").join(fragments)
  856. async def read_data_frame(self, max_size: int | None) -> Frame | None:
  857. """
  858. Read a single data frame from the connection.
  859. Process control frames received before the next data frame.
  860. Return :obj:`None` if a close frame is encountered before any data frame.
  861. """
  862. # 6.2. Receiving Data
  863. while True:
  864. frame = await self.read_frame(max_size)
  865. # 5.5. Control Frames
  866. if frame.opcode == OP_CLOSE:
  867. # 7.1.5. The WebSocket Connection Close Code
  868. # 7.1.6. The WebSocket Connection Close Reason
  869. self.close_rcvd = Close.parse(frame.data)
  870. if self.close_sent is not None:
  871. self.close_rcvd_then_sent = False
  872. try:
  873. # Echo the original data instead of re-serializing it with
  874. # Close.serialize() because that fails when the close frame
  875. # is empty and Close.parse() synthesizes a 1005 close code.
  876. await self.write_close_frame(self.close_rcvd, frame.data)
  877. except ConnectionClosed:
  878. # Connection closed before we could echo the close frame.
  879. pass
  880. return None
  881. elif frame.opcode == OP_PING:
  882. # Answer pings, unless connection is CLOSING.
  883. if self.state is State.OPEN:
  884. try:
  885. await self.pong(frame.data)
  886. except ConnectionClosed:
  887. # Connection closed while draining write buffer.
  888. pass
  889. elif frame.opcode == OP_PONG:
  890. if frame.data in self.pings:
  891. pong_timestamp = time.perf_counter()
  892. # Sending a pong for only the most recent ping is legal.
  893. # Acknowledge all previous pings too in that case.
  894. ping_id = None
  895. ping_ids = []
  896. for ping_id, (pong_waiter, ping_timestamp) in self.pings.items():
  897. ping_ids.append(ping_id)
  898. if not pong_waiter.done():
  899. pong_waiter.set_result(pong_timestamp - ping_timestamp)
  900. if ping_id == frame.data:
  901. self.latency = pong_timestamp - ping_timestamp
  902. break
  903. else:
  904. raise AssertionError("solicited pong not found in pings")
  905. # Remove acknowledged pings from self.pings.
  906. for ping_id in ping_ids:
  907. del self.pings[ping_id]
  908. # 5.6. Data Frames
  909. else:
  910. return frame
  911. async def read_frame(self, max_size: int | None) -> Frame:
  912. """
  913. Read a single frame from the connection.
  914. """
  915. frame = await Frame.read(
  916. self.reader.readexactly,
  917. mask=not self.is_client,
  918. max_size=max_size,
  919. extensions=self.extensions,
  920. )
  921. if self.debug:
  922. self.logger.debug("< %s", frame)
  923. return frame
  924. def write_frame_sync(self, fin: bool, opcode: int, data: BytesLike) -> None:
  925. frame = Frame(fin, Opcode(opcode), data)
  926. if self.debug:
  927. self.logger.debug("> %s", frame)
  928. frame.write(
  929. self.transport.write,
  930. mask=self.is_client,
  931. extensions=self.extensions,
  932. )
  933. async def drain(self) -> None:
  934. try:
  935. # Handle flow control automatically.
  936. await self._drain()
  937. except ConnectionError:
  938. # Terminate the connection if the socket died.
  939. self.fail_connection()
  940. # Wait until the connection is closed to raise ConnectionClosed
  941. # with the correct code and reason.
  942. await self.ensure_open()
  943. async def write_frame(
  944. self, fin: bool, opcode: int, data: BytesLike, *, _state: int = State.OPEN
  945. ) -> None:
  946. # Defensive assertion for protocol compliance.
  947. if self.state is not _state: # pragma: no cover
  948. raise InvalidState(
  949. f"Cannot write to a WebSocket in the {self.state.name} state"
  950. )
  951. self.write_frame_sync(fin, opcode, data)
  952. await self.drain()
  953. async def write_close_frame(
  954. self, close: Close, data: BytesLike | None = None
  955. ) -> None:
  956. """
  957. Write a close frame if and only if the connection state is OPEN.
  958. This dedicated coroutine must be used for writing close frames to
  959. ensure that at most one close frame is sent on a given connection.
  960. """
  961. # Test and set the connection state before sending the close frame to
  962. # avoid sending two frames in case of concurrent calls.
  963. if self.state is State.OPEN:
  964. # 7.1.3. The WebSocket Closing Handshake is Started
  965. self.state = State.CLOSING
  966. if self.debug:
  967. self.logger.debug("= connection is CLOSING")
  968. self.close_sent = close
  969. if self.close_rcvd is not None:
  970. self.close_rcvd_then_sent = True
  971. if data is None:
  972. data = close.serialize()
  973. # 7.1.2. Start the WebSocket Closing Handshake
  974. await self.write_frame(True, OP_CLOSE, data, _state=State.CLOSING)
  975. async def keepalive_ping(self) -> None:
  976. """
  977. Send a Ping frame and wait for a Pong frame at regular intervals.
  978. This coroutine exits when the connection terminates and one of the
  979. following happens:
  980. - :meth:`ping` raises :exc:`ConnectionClosed`, or
  981. - :meth:`close_connection` cancels :attr:`keepalive_ping_task`.
  982. """
  983. if self.ping_interval is None:
  984. return
  985. try:
  986. while True:
  987. await asyncio.sleep(self.ping_interval)
  988. if self.debug:
  989. self.logger.debug("% sending keepalive ping")
  990. pong_waiter = await self.ping()
  991. if self.ping_timeout is not None:
  992. try:
  993. async with asyncio_timeout(self.ping_timeout):
  994. # Raises CancelledError if the connection is closed,
  995. # when close_connection() cancels keepalive_ping().
  996. # Raises ConnectionClosed if the connection is lost,
  997. # when connection_lost() calls abort_pings().
  998. await pong_waiter
  999. if self.debug:
  1000. self.logger.debug("% received keepalive pong")
  1001. except asyncio.TimeoutError:
  1002. if self.debug:
  1003. self.logger.debug("- timed out waiting for keepalive pong")
  1004. self.fail_connection(
  1005. CloseCode.INTERNAL_ERROR,
  1006. "keepalive ping timeout",
  1007. )
  1008. break
  1009. except ConnectionClosed:
  1010. pass
  1011. except Exception:
  1012. self.logger.error("keepalive ping failed", exc_info=True)
  1013. async def close_connection(self) -> None:
  1014. """
  1015. 7.1.1. Close the WebSocket Connection
  1016. When the opening handshake succeeds, :meth:`connection_open` starts
  1017. this coroutine in a task. It waits for the data transfer phase to
  1018. complete then it closes the TCP connection cleanly.
  1019. When the opening handshake fails, :meth:`fail_connection` does the
  1020. same. There's no data transfer phase in that case.
  1021. """
  1022. try:
  1023. # Wait for the data transfer phase to complete.
  1024. if hasattr(self, "transfer_data_task"):
  1025. try:
  1026. await self.transfer_data_task
  1027. except asyncio.CancelledError:
  1028. pass
  1029. # Cancel the keepalive ping task.
  1030. if hasattr(self, "keepalive_ping_task"):
  1031. self.keepalive_ping_task.cancel()
  1032. # A client should wait for a TCP close from the server.
  1033. if self.is_client and hasattr(self, "transfer_data_task"):
  1034. if await self.wait_for_connection_lost():
  1035. return
  1036. if self.debug:
  1037. self.logger.debug("- timed out waiting for TCP close")
  1038. # Half-close the TCP connection if possible (when there's no TLS).
  1039. if self.transport.can_write_eof():
  1040. if self.debug:
  1041. self.logger.debug("x half-closing TCP connection")
  1042. # write_eof() doesn't document which exceptions it raises.
  1043. # "[Errno 107] Transport endpoint is not connected" happens
  1044. # but it isn't completely clear under which circumstances.
  1045. # uvloop can raise RuntimeError here.
  1046. try:
  1047. self.transport.write_eof()
  1048. except (OSError, RuntimeError): # pragma: no cover
  1049. pass
  1050. if await self.wait_for_connection_lost():
  1051. return
  1052. if self.debug:
  1053. self.logger.debug("- timed out waiting for TCP close")
  1054. finally:
  1055. # The try/finally ensures that the transport never remains open,
  1056. # even if this coroutine is canceled (for example).
  1057. await self.close_transport()
  1058. async def close_transport(self) -> None:
  1059. """
  1060. Close the TCP connection.
  1061. """
  1062. # If connection_lost() was called, the TCP connection is closed.
  1063. # However, if TLS is enabled, the transport still needs closing.
  1064. # Else asyncio complains: ResourceWarning: unclosed transport.
  1065. if self.connection_lost_waiter.done() and self.transport.is_closing():
  1066. return
  1067. # Close the TCP connection. Buffers are flushed asynchronously.
  1068. if self.debug:
  1069. self.logger.debug("x closing TCP connection")
  1070. self.transport.close()
  1071. if await self.wait_for_connection_lost():
  1072. return
  1073. if self.debug:
  1074. self.logger.debug("- timed out waiting for TCP close")
  1075. # Abort the TCP connection. Buffers are discarded.
  1076. if self.debug:
  1077. self.logger.debug("x aborting TCP connection")
  1078. self.transport.abort()
  1079. # connection_lost() is called quickly after aborting.
  1080. await self.wait_for_connection_lost()
  1081. async def wait_for_connection_lost(self) -> bool:
  1082. """
  1083. Wait until the TCP connection is closed or ``self.close_timeout`` elapses.
  1084. Return :obj:`True` if the connection is closed and :obj:`False`
  1085. otherwise.
  1086. """
  1087. if not self.connection_lost_waiter.done():
  1088. try:
  1089. async with asyncio_timeout(self.close_timeout):
  1090. await asyncio.shield(self.connection_lost_waiter)
  1091. except asyncio.TimeoutError:
  1092. pass
  1093. # Re-check self.connection_lost_waiter.done() synchronously because
  1094. # connection_lost() could run between the moment the timeout occurs
  1095. # and the moment this coroutine resumes running.
  1096. return self.connection_lost_waiter.done()
  1097. def fail_connection(
  1098. self,
  1099. code: int = CloseCode.ABNORMAL_CLOSURE,
  1100. reason: str = "",
  1101. ) -> None:
  1102. """
  1103. 7.1.7. Fail the WebSocket Connection
  1104. This requires:
  1105. 1. Stopping all processing of incoming data, which means canceling
  1106. :attr:`transfer_data_task`. The close code will be 1006 unless a
  1107. close frame was received earlier.
  1108. 2. Sending a close frame with an appropriate code if the opening
  1109. handshake succeeded and the other side is likely to process it.
  1110. 3. Closing the connection. :meth:`close_connection` takes care of
  1111. this once :attr:`transfer_data_task` exits after being canceled.
  1112. (The specification describes these steps in the opposite order.)
  1113. """
  1114. if self.debug:
  1115. self.logger.debug("! failing connection with code %d", code)
  1116. # Cancel transfer_data_task if the opening handshake succeeded.
  1117. # cancel() is idempotent and ignored if the task is done already.
  1118. if hasattr(self, "transfer_data_task"):
  1119. self.transfer_data_task.cancel()
  1120. # Send a close frame when the state is OPEN (a close frame was already
  1121. # sent if it's CLOSING), except when failing the connection because of
  1122. # an error reading from or writing to the network.
  1123. # Don't send a close frame if the connection is broken.
  1124. if code != CloseCode.ABNORMAL_CLOSURE and self.state is State.OPEN:
  1125. close = Close(code, reason)
  1126. # Write the close frame without draining the write buffer.
  1127. # Keeping fail_connection() synchronous guarantees it can't
  1128. # get stuck and simplifies the implementation of the callers.
  1129. # Not drainig the write buffer is acceptable in this context.
  1130. # This duplicates a few lines of code from write_close_frame().
  1131. self.state = State.CLOSING
  1132. if self.debug:
  1133. self.logger.debug("= connection is CLOSING")
  1134. # If self.close_rcvd was set, the connection state would be
  1135. # CLOSING. Therefore self.close_rcvd isn't set and we don't
  1136. # have to set self.close_rcvd_then_sent.
  1137. assert self.close_rcvd is None
  1138. self.close_sent = close
  1139. self.write_frame_sync(True, OP_CLOSE, close.serialize())
  1140. # Start close_connection_task if the opening handshake didn't succeed.
  1141. if not hasattr(self, "close_connection_task"):
  1142. self.close_connection_task = self.loop.create_task(self.close_connection())
  1143. def abort_pings(self) -> None:
  1144. """
  1145. Raise ConnectionClosed in pending keepalive pings.
  1146. They'll never receive a pong once the connection is closed.
  1147. """
  1148. assert self.state is State.CLOSED
  1149. exc = self.connection_closed_exc()
  1150. for pong_waiter, _ping_timestamp in self.pings.values():
  1151. pong_waiter.set_exception(exc)
  1152. # If the exception is never retrieved, it will be logged when ping
  1153. # is garbage-collected. This is confusing for users.
  1154. # Given that ping is done (with an exception), canceling it does
  1155. # nothing, but it prevents logging the exception.
  1156. pong_waiter.cancel()
  1157. # asyncio.Protocol methods
  1158. def connection_made(self, transport: asyncio.BaseTransport) -> None:
  1159. """
  1160. Configure write buffer limits.
  1161. The high-water limit is defined by ``self.write_limit``.
  1162. The low-water limit currently defaults to ``self.write_limit // 4`` in
  1163. :meth:`~asyncio.WriteTransport.set_write_buffer_limits`, which should
  1164. be all right for reasonable use cases of this library.
  1165. This is the earliest point where we can get hold of the transport,
  1166. which means it's the best point for configuring it.
  1167. """
  1168. transport = cast(asyncio.Transport, transport)
  1169. transport.set_write_buffer_limits(self.write_limit)
  1170. self.transport = transport
  1171. # Copied from asyncio.StreamReaderProtocol
  1172. self.reader.set_transport(transport)
  1173. def connection_lost(self, exc: Exception | None) -> None:
  1174. """
  1175. 7.1.4. The WebSocket Connection is Closed.
  1176. """
  1177. self.state = State.CLOSED
  1178. if self.debug:
  1179. self.logger.debug("= connection is CLOSED")
  1180. self.abort_pings()
  1181. # If self.connection_lost_waiter isn't pending, that's a bug, because:
  1182. # - it's set only here in connection_lost() which is called only once;
  1183. # - it must never be canceled.
  1184. self.connection_lost_waiter.set_result(None)
  1185. if True: # pragma: no cover
  1186. # Copied from asyncio.StreamReaderProtocol
  1187. if self.reader is not None:
  1188. if exc is None:
  1189. self.reader.feed_eof()
  1190. else:
  1191. self.reader.set_exception(exc)
  1192. # Copied from asyncio.FlowControlMixin
  1193. # Wake up the writer if currently paused.
  1194. if not self._paused:
  1195. return
  1196. waiter = self._drain_waiter
  1197. if waiter is None:
  1198. return
  1199. self._drain_waiter = None
  1200. if waiter.done():
  1201. return
  1202. if exc is None:
  1203. waiter.set_result(None)
  1204. else:
  1205. waiter.set_exception(exc)
  1206. def pause_writing(self) -> None: # pragma: no cover
  1207. assert not self._paused
  1208. self._paused = True
  1209. def resume_writing(self) -> None: # pragma: no cover
  1210. assert self._paused
  1211. self._paused = False
  1212. waiter = self._drain_waiter
  1213. if waiter is not None:
  1214. self._drain_waiter = None
  1215. if not waiter.done():
  1216. waiter.set_result(None)
  1217. def data_received(self, data: bytes) -> None:
  1218. self.reader.feed_data(data)
  1219. def eof_received(self) -> None:
  1220. """
  1221. Close the transport after receiving EOF.
  1222. The WebSocket protocol has its own closing handshake: endpoints close
  1223. the TCP or TLS connection after sending and receiving a close frame.
  1224. As a consequence, they never need to write after receiving EOF, so
  1225. there's no reason to keep the transport open by returning :obj:`True`.
  1226. Besides, that doesn't work on TLS connections.
  1227. """
  1228. self.reader.feed_eof()
  1229. # broadcast() is defined in the protocol module even though it's primarily
  1230. # used by servers and documented in the server module because it works with
  1231. # client connections too and because it's easier to test together with the
  1232. # WebSocketCommonProtocol class.
  1233. def broadcast(
  1234. websockets: Iterable[WebSocketCommonProtocol],
  1235. message: DataLike,
  1236. raise_exceptions: bool = False,
  1237. ) -> None:
  1238. """
  1239. Broadcast a message to several WebSocket connections.
  1240. A string (:class:`str`) is sent as a Text_ frame. A bytestring or bytes-like
  1241. object (:class:`bytes`, :class:`bytearray`, or :class:`memoryview`) is sent
  1242. as a Binary_ frame.
  1243. .. _Text: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  1244. .. _Binary: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  1245. :func:`broadcast` pushes the message synchronously to all connections even
  1246. if their write buffers are overflowing. There's no backpressure.
  1247. If you broadcast messages faster than a connection can handle them, messages
  1248. will pile up in its write buffer until the connection times out. Keep
  1249. ``ping_interval`` and ``ping_timeout`` low to prevent excessive memory usage
  1250. from slow connections.
  1251. Unlike :meth:`~websockets.legacy.protocol.WebSocketCommonProtocol.send`,
  1252. :func:`broadcast` doesn't support sending fragmented messages. Indeed,
  1253. fragmentation is useful for sending large messages without buffering them in
  1254. memory, while :func:`broadcast` buffers one copy per connection as fast as
  1255. possible.
  1256. :func:`broadcast` skips connections that aren't open in order to avoid
  1257. errors on connections where the closing handshake is in progress.
  1258. :func:`broadcast` ignores failures to write the message on some connections.
  1259. It continues writing to other connections. On Python 3.11 and above, you may
  1260. set ``raise_exceptions`` to :obj:`True` to record failures and raise all
  1261. exceptions in a :pep:`654` :exc:`ExceptionGroup`.
  1262. While :func:`broadcast` makes more sense for servers, it works identically
  1263. with clients, if you have a use case for opening connections to many servers
  1264. and broadcasting a message to them.
  1265. Args:
  1266. websockets: WebSocket connections to which the message will be sent.
  1267. message: Message to send.
  1268. raise_exceptions: Whether to raise an exception in case of failures.
  1269. Raises:
  1270. TypeError: If ``message`` doesn't have a supported type.
  1271. """
  1272. if not isinstance(message, (str, bytes, bytearray, memoryview)):
  1273. raise TypeError("data must be str or bytes-like")
  1274. if raise_exceptions:
  1275. if sys.version_info[:2] < (3, 11): # pragma: no cover
  1276. raise ValueError("raise_exceptions requires at least Python 3.11")
  1277. exceptions = []
  1278. opcode, data = prepare_data(message)
  1279. for websocket in websockets:
  1280. if websocket.state is not State.OPEN:
  1281. continue
  1282. if websocket._fragmented_message_waiter is not None:
  1283. if raise_exceptions:
  1284. exception = RuntimeError("sending a fragmented message")
  1285. exceptions.append(exception)
  1286. else:
  1287. websocket.logger.warning(
  1288. "skipped broadcast: sending a fragmented message",
  1289. )
  1290. continue
  1291. try:
  1292. websocket.write_frame_sync(True, opcode, data)
  1293. except Exception as write_exception:
  1294. if raise_exceptions:
  1295. exception = RuntimeError("failed to write message")
  1296. exception.__cause__ = write_exception
  1297. exceptions.append(exception)
  1298. else:
  1299. websocket.logger.warning(
  1300. "skipped broadcast: failed to write message: %s",
  1301. traceback.format_exception_only(write_exception)[0].strip(),
  1302. )
  1303. if raise_exceptions and exceptions:
  1304. raise ExceptionGroup("skipped broadcast", exceptions)
  1305. # Pretend that broadcast is actually defined in the server module.
  1306. broadcast.__module__ = "websockets.legacy.server"