Funktionierender Prototyp des Serious Games zur Vermittlung von Wissen zu Software-Engineering-Arbeitsmodellen.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

protocol.py 62KB

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