connection.py 41 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078
  1. from __future__ import annotations
  2. import contextlib
  3. import logging
  4. import random
  5. import socket
  6. import struct
  7. import threading
  8. import time
  9. import uuid
  10. from collections.abc import Iterable, Iterator, Mapping
  11. from types import TracebackType
  12. from typing import Any, Literal, overload
  13. from ..exceptions import (
  14. ConcurrencyError,
  15. ConnectionClosed,
  16. ConnectionClosedOK,
  17. ProtocolError,
  18. )
  19. from ..frames import DATA_OPCODES, CloseCode, Frame, Opcode
  20. from ..http11 import Request, Response
  21. from ..protocol import CLOSED, OPEN, Event, Protocol, State
  22. from ..typing import BytesLike, Data, DataLike, LoggerLike, Subprotocol
  23. from .messages import Assembler
  24. from .utils import Deadline
  25. __all__ = ["Connection"]
  26. class Connection:
  27. """
  28. :mod:`threading` implementation of a WebSocket connection.
  29. :class:`Connection` provides APIs shared between WebSocket servers and
  30. clients.
  31. You shouldn't use it directly. Instead, use
  32. :class:`~websockets.sync.client.ClientConnection` or
  33. :class:`~websockets.sync.server.ServerConnection`.
  34. """
  35. recv_bufsize = 65536
  36. def __init__(
  37. self,
  38. socket: socket.socket,
  39. protocol: Protocol,
  40. *,
  41. ping_interval: float | None = 20,
  42. ping_timeout: float | None = 20,
  43. close_timeout: float | None = 10,
  44. max_queue: int | None | tuple[int | None, int | None] = 16,
  45. ) -> None:
  46. self.socket = socket
  47. self.protocol = protocol
  48. self.ping_interval = ping_interval
  49. self.ping_timeout = ping_timeout
  50. self.close_timeout = close_timeout
  51. if isinstance(max_queue, int) or max_queue is None:
  52. max_queue_high, max_queue_low = max_queue, None
  53. else:
  54. max_queue_high, max_queue_low = max_queue
  55. # Inject reference to this instance in the protocol's logger.
  56. self.protocol.logger = logging.LoggerAdapter(
  57. self.protocol.logger,
  58. {"websocket": self},
  59. )
  60. # Copy attributes from the protocol for convenience.
  61. self.id: uuid.UUID = self.protocol.id
  62. """Unique identifier of the connection. Useful in logs."""
  63. self.logger: LoggerLike = self.protocol.logger
  64. """Logger for this connection."""
  65. self.debug = self.protocol.debug
  66. # HTTP handshake request and response.
  67. self.request: Request | None = None
  68. """Opening handshake request."""
  69. self.response: Response | None = None
  70. """Opening handshake response."""
  71. # Mutex serializing interactions with the protocol.
  72. self.protocol_mutex = threading.Lock()
  73. # Lock stopping reads when the assembler buffer is full.
  74. self.recv_flow_control = threading.Lock()
  75. # Assembler turning frames into messages and serializing reads.
  76. self.recv_messages = Assembler(
  77. max_queue_high,
  78. max_queue_low,
  79. pause=self.recv_flow_control.acquire,
  80. resume=self.recv_flow_control.release,
  81. )
  82. # Deadline for the closing handshake.
  83. self.close_deadline: Deadline | None = None
  84. # Whether we are busy sending a fragmented message.
  85. self.send_in_progress = False
  86. # Mapping of ping IDs to pong waiters, in chronological order.
  87. self.pending_pings: dict[bytes, tuple[threading.Event, float, bool]] = {}
  88. self.latency: float = 0.0
  89. """
  90. Latency of the connection, in seconds.
  91. Latency is defined as the round-trip time of the connection. It is
  92. measured by sending a Ping frame and waiting for a matching Pong frame.
  93. Before the first measurement, :attr:`latency` is ``0.0``.
  94. By default, websockets enables a :ref:`keepalive <keepalive>` mechanism
  95. that sends Ping frames automatically at regular intervals. You can also
  96. send Ping frames and measure latency with :meth:`ping`.
  97. """
  98. # Thread that sends keepalive pings. None when ping_interval is None.
  99. self.keepalive_thread: threading.Thread | None = None
  100. # Exception raised while reading from the connection, to be chained to
  101. # ConnectionClosed in order to show why the TCP connection dropped.
  102. self.recv_exc: BaseException | None = None
  103. # Receiving events from the socket. This thread is marked as daemon to
  104. # allow creating a connection in a non-daemon thread and using it in a
  105. # daemon thread. This mustn't prevent the interpreter from exiting.
  106. self.recv_events_thread = threading.Thread(
  107. target=self.recv_events,
  108. daemon=True,
  109. )
  110. # Start recv_events only after all attributes are initialized.
  111. self.recv_events_thread.start()
  112. # Public attributes
  113. @property
  114. def local_address(self) -> Any:
  115. """
  116. Local address of the connection.
  117. For IPv4 connections, this is a ``(host, port)`` tuple.
  118. The format of the address depends on the address family.
  119. See :meth:`~socket.socket.getsockname`.
  120. """
  121. return self.socket.getsockname()
  122. @property
  123. def remote_address(self) -> Any:
  124. """
  125. Remote address of the connection.
  126. For IPv4 connections, this is a ``(host, port)`` tuple.
  127. The format of the address depends on the address family.
  128. See :meth:`~socket.socket.getpeername`.
  129. """
  130. return self.socket.getpeername()
  131. @property
  132. def state(self) -> State:
  133. """
  134. State of the WebSocket connection, defined in :rfc:`6455`.
  135. This attribute is provided for completeness. Typical applications
  136. shouldn't check its value. Instead, they should call :meth:`~recv` or
  137. :meth:`send` and handle :exc:`~websockets.exceptions.ConnectionClosed`
  138. exceptions.
  139. """
  140. return self.protocol.state
  141. @property
  142. def subprotocol(self) -> Subprotocol | None:
  143. """
  144. Subprotocol negotiated during the opening handshake.
  145. :obj:`None` if no subprotocol was negotiated.
  146. """
  147. return self.protocol.subprotocol
  148. @property
  149. def close_code(self) -> int | None:
  150. """
  151. State of the WebSocket connection, defined in :rfc:`6455`.
  152. This attribute is provided for completeness. Typical applications
  153. shouldn't check its value. Instead, they should inspect attributes
  154. of :exc:`~websockets.exceptions.ConnectionClosed` exceptions.
  155. """
  156. return self.protocol.close_code
  157. @property
  158. def close_reason(self) -> str | None:
  159. """
  160. State of the WebSocket connection, defined in :rfc:`6455`.
  161. This attribute is provided for completeness. Typical applications
  162. shouldn't check its value. Instead, they should inspect attributes
  163. of :exc:`~websockets.exceptions.ConnectionClosed` exceptions.
  164. """
  165. return self.protocol.close_reason
  166. # Public methods
  167. def __enter__(self) -> Connection:
  168. return self
  169. def __exit__(
  170. self,
  171. exc_type: type[BaseException] | None,
  172. exc_value: BaseException | None,
  173. traceback: TracebackType | None,
  174. ) -> None:
  175. if exc_type is None:
  176. self.close()
  177. else:
  178. self.close(CloseCode.INTERNAL_ERROR)
  179. def __iter__(self) -> Iterator[Data]:
  180. """
  181. Iterate on incoming messages.
  182. The iterator calls :meth:`recv` and yields messages in an infinite loop.
  183. It exits when the connection is closed normally. It raises a
  184. :exc:`~websockets.exceptions.ConnectionClosedError` exception after a
  185. protocol error or a network failure.
  186. """
  187. try:
  188. while True:
  189. yield self.recv()
  190. except ConnectionClosedOK:
  191. return
  192. # This overload structure is required to avoid the error:
  193. # "parameter without a default follows parameter with a default"
  194. @overload
  195. def recv(self, timeout: float | None, decode: Literal[True]) -> str: ...
  196. @overload
  197. def recv(self, timeout: float | None, decode: Literal[False]) -> bytes: ...
  198. @overload
  199. def recv(self, timeout: float | None = None, *, decode: Literal[True]) -> str: ...
  200. @overload
  201. def recv(
  202. self, timeout: float | None = None, *, decode: Literal[False]
  203. ) -> bytes: ...
  204. @overload
  205. def recv(
  206. self, timeout: float | None = None, decode: bool | None = None
  207. ) -> Data: ...
  208. def recv(self, timeout: float | None = None, decode: bool | None = None) -> Data:
  209. """
  210. Receive the next message.
  211. When the connection is closed, :meth:`recv` raises
  212. :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it raises
  213. :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal closure
  214. and :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol
  215. error or a network failure. This is how you detect the end of the
  216. message stream.
  217. If ``timeout`` is :obj:`None`, block until a message is received. If
  218. ``timeout`` is set, wait up to ``timeout`` seconds for a message to be
  219. received and return it, else raise :exc:`TimeoutError`. If ``timeout``
  220. is ``0`` or negative, check if a message has been received already and
  221. return it, else raise :exc:`TimeoutError`.
  222. When the message is fragmented, :meth:`recv` waits until all fragments
  223. are received, reassembles them, and returns the whole message.
  224. Args:
  225. timeout: Timeout for receiving a message in seconds.
  226. decode: Set this flag to override the default behavior of returning
  227. :class:`str` or :class:`bytes`. See below for details.
  228. Returns:
  229. A string (:class:`str`) for a Text_ frame or a bytestring
  230. (:class:`bytes`) for a Binary_ frame.
  231. .. _Text: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  232. .. _Binary: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  233. You may override this behavior with the ``decode`` argument:
  234. * Set ``decode=False`` to disable UTF-8 decoding of Text_ frames and
  235. return a bytestring (:class:`bytes`). This improves performance
  236. when decoding isn't needed, for example if the message contains
  237. JSON and you're using a JSON library that expects a bytestring.
  238. * Set ``decode=True`` to force UTF-8 decoding of Binary_ frames and
  239. return strings (:class:`str`). This may be useful for servers that
  240. send binary frames instead of text frames.
  241. Raises:
  242. ConnectionClosed: When the connection is closed.
  243. ConcurrencyError: If two threads call :meth:`recv` or
  244. :meth:`recv_streaming` concurrently.
  245. """
  246. try:
  247. return self.recv_messages.get(timeout, decode)
  248. except EOFError:
  249. pass
  250. # fallthrough
  251. except ConcurrencyError:
  252. raise ConcurrencyError(
  253. "cannot call recv while another thread "
  254. "is already running recv or recv_streaming"
  255. ) from None
  256. except UnicodeDecodeError as exc:
  257. with self.send_context():
  258. self.protocol.fail(
  259. CloseCode.INVALID_DATA,
  260. f"{exc.reason} at position {exc.start}",
  261. )
  262. # fallthrough
  263. # Wait for the protocol state to be CLOSED before accessing close_exc.
  264. self.recv_events_thread.join()
  265. raise self.protocol.close_exc from self.recv_exc
  266. @overload
  267. def recv_streaming(self, decode: Literal[True]) -> Iterator[str]: ...
  268. @overload
  269. def recv_streaming(self, decode: Literal[False]) -> Iterator[bytes]: ...
  270. @overload
  271. def recv_streaming(self, decode: bool | None = None) -> Iterator[Data]: ...
  272. def recv_streaming(self, decode: bool | None = None) -> Iterator[Data]:
  273. """
  274. Receive the next message frame by frame.
  275. This method is designed for receiving fragmented messages. It returns an
  276. iterator that yields each fragment as it is received. This iterator must
  277. be fully consumed. Else, future calls to :meth:`recv` or
  278. :meth:`recv_streaming` will raise
  279. :exc:`~websockets.exceptions.ConcurrencyError`, making the connection
  280. unusable.
  281. :meth:`recv_streaming` raises the same exceptions as :meth:`recv`.
  282. Args:
  283. decode: Set this flag to override the default behavior of returning
  284. :class:`str` or :class:`bytes`. See below for details.
  285. Returns:
  286. An iterator of strings (:class:`str`) for a Text_ frame or
  287. bytestrings (:class:`bytes`) for a Binary_ frame.
  288. .. _Text: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  289. .. _Binary: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  290. You may override this behavior with the ``decode`` argument:
  291. * Set ``decode=False`` to disable UTF-8 decoding of Text_ frames and
  292. yield bytestrings (:class:`bytes`). This improves performance
  293. when decoding isn't needed.
  294. * Set ``decode=True`` to force UTF-8 decoding of Binary_ frames and
  295. yield strings (:class:`str`). This may be useful for servers that
  296. send binary frames instead of text frames.
  297. Raises:
  298. ConnectionClosed: When the connection is closed.
  299. ConcurrencyError: If two threads call :meth:`recv` or
  300. :meth:`recv_streaming` concurrently.
  301. """
  302. try:
  303. yield from self.recv_messages.get_iter(decode)
  304. return
  305. except EOFError:
  306. pass
  307. # fallthrough
  308. except ConcurrencyError:
  309. raise ConcurrencyError(
  310. "cannot call recv_streaming while another thread "
  311. "is already running recv or recv_streaming"
  312. ) from None
  313. except UnicodeDecodeError as exc:
  314. with self.send_context():
  315. self.protocol.fail(
  316. CloseCode.INVALID_DATA,
  317. f"{exc.reason} at position {exc.start}",
  318. )
  319. # fallthrough
  320. # Wait for the protocol state to be CLOSED before accessing close_exc.
  321. self.recv_events_thread.join()
  322. raise self.protocol.close_exc from self.recv_exc
  323. def send(
  324. self,
  325. message: DataLike | Iterable[DataLike],
  326. text: bool | None = None,
  327. ) -> None:
  328. """
  329. Send a message.
  330. A string (:class:`str`) is sent as a Text_ frame. A bytestring or
  331. bytes-like object (:class:`bytes`, :class:`bytearray`, or
  332. :class:`memoryview`) is sent as a Binary_ frame.
  333. .. _Text: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  334. .. _Binary: https://datatracker.ietf.org/doc/html/rfc6455#section-5.6
  335. You may override this behavior with the ``text`` argument:
  336. * Set ``text=True`` to send an UTF-8 bytestring or bytes-like object
  337. (:class:`bytes`, :class:`bytearray`, or :class:`memoryview`) in a
  338. Text_ frame. This improves performance when the message is already
  339. UTF-8 encoded, for example if the message contains JSON and you're
  340. using a JSON library that produces a bytestring.
  341. * Set ``text=False`` to send a string (:class:`str`) in a Binary_
  342. frame. This may be useful for servers that expect binary frames
  343. instead of text frames.
  344. :meth:`send` also accepts an iterable of strings, bytestrings, or
  345. bytes-like objects to enable fragmentation_. Each item is treated as a
  346. message fragment and sent in its own frame. All items must be of the
  347. same type, or else :meth:`send` will raise a :exc:`TypeError` and the
  348. connection will be closed.
  349. .. _fragmentation: https://datatracker.ietf.org/doc/html/rfc6455#section-5.4
  350. :meth:`send` rejects dict-like objects because this is often an error.
  351. (If you really want to send the keys of a dict-like object as fragments,
  352. call its :meth:`~dict.keys` method and pass the result to :meth:`send`.)
  353. When the connection is closed, :meth:`send` raises
  354. :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it
  355. raises :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal
  356. connection closure and
  357. :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol
  358. error or a network failure.
  359. Args:
  360. message: Message to send.
  361. Raises:
  362. ConnectionClosed: When the connection is closed.
  363. ConcurrencyError: If the connection is sending a fragmented message.
  364. TypeError: If ``message`` doesn't have a supported type.
  365. """
  366. # Unfragmented message -- this case must be handled first because
  367. # strings and bytes-like objects are iterable.
  368. if isinstance(message, str):
  369. with self.send_context():
  370. if self.send_in_progress:
  371. raise ConcurrencyError(
  372. "cannot call send while another thread is already running send"
  373. )
  374. if text is False:
  375. self.protocol.send_binary(message.encode())
  376. else:
  377. self.protocol.send_text(message.encode())
  378. elif isinstance(message, BytesLike):
  379. with self.send_context():
  380. if self.send_in_progress:
  381. raise ConcurrencyError(
  382. "cannot call send while another thread is already running send"
  383. )
  384. if text is True:
  385. self.protocol.send_text(message)
  386. else:
  387. self.protocol.send_binary(message)
  388. # Catch a common mistake -- passing a dict to send().
  389. elif isinstance(message, Mapping):
  390. raise TypeError("data is a dict-like object")
  391. # Fragmented message -- regular iterator.
  392. elif isinstance(message, Iterable):
  393. chunks = iter(message)
  394. try:
  395. chunk = next(chunks)
  396. except StopIteration:
  397. return
  398. try:
  399. # First fragment.
  400. if isinstance(chunk, str):
  401. with self.send_context():
  402. if self.send_in_progress:
  403. raise ConcurrencyError(
  404. "cannot call send while another thread "
  405. "is already running send"
  406. )
  407. self.send_in_progress = True
  408. if text is False:
  409. self.protocol.send_binary(chunk.encode(), fin=False)
  410. else:
  411. self.protocol.send_text(chunk.encode(), fin=False)
  412. encode = True
  413. elif isinstance(chunk, BytesLike):
  414. with self.send_context():
  415. if self.send_in_progress:
  416. raise ConcurrencyError(
  417. "cannot call send while another thread "
  418. "is already running send"
  419. )
  420. self.send_in_progress = True
  421. if text is True:
  422. self.protocol.send_text(chunk, fin=False)
  423. else:
  424. self.protocol.send_binary(chunk, fin=False)
  425. encode = False
  426. else:
  427. raise TypeError("iterable must contain bytes or str")
  428. # Other fragments
  429. for chunk in chunks:
  430. if isinstance(chunk, str) and encode:
  431. with self.send_context():
  432. assert self.send_in_progress
  433. self.protocol.send_continuation(chunk.encode(), fin=False)
  434. elif isinstance(chunk, BytesLike) and not encode:
  435. with self.send_context():
  436. assert self.send_in_progress
  437. self.protocol.send_continuation(chunk, fin=False)
  438. else:
  439. raise TypeError("iterable must contain uniform types")
  440. # Final fragment.
  441. with self.send_context():
  442. self.protocol.send_continuation(b"", fin=True)
  443. self.send_in_progress = False
  444. except ConcurrencyError:
  445. # We didn't start sending a fragmented message.
  446. # The connection is still usable.
  447. raise
  448. except Exception:
  449. # We're half-way through a fragmented message and we can't
  450. # complete it. This makes the connection unusable.
  451. with self.send_context():
  452. self.protocol.fail(
  453. CloseCode.INTERNAL_ERROR,
  454. "error in fragmented message",
  455. )
  456. raise
  457. else:
  458. raise TypeError("data must be str, bytes, or iterable")
  459. def close(
  460. self,
  461. code: CloseCode | int = CloseCode.NORMAL_CLOSURE,
  462. reason: str = "",
  463. ) -> None:
  464. """
  465. Perform the closing handshake.
  466. :meth:`close` waits for the other end to complete the handshake and
  467. for the TCP connection to terminate.
  468. :meth:`close` is idempotent: it doesn't do anything once the
  469. connection is closed.
  470. Args:
  471. code: WebSocket close code.
  472. reason: WebSocket close reason.
  473. """
  474. try:
  475. # The context manager takes care of waiting for the TCP connection
  476. # to terminate after calling a method that sends a close frame.
  477. with self.send_context():
  478. if self.send_in_progress:
  479. self.protocol.fail(
  480. CloseCode.INTERNAL_ERROR,
  481. "close during fragmented message",
  482. )
  483. else:
  484. self.protocol.send_close(code, reason)
  485. except ConnectionClosed:
  486. # Ignore ConnectionClosed exceptions raised from send_context().
  487. # They mean that the connection is closed, which was the goal.
  488. pass
  489. def ping(
  490. self,
  491. data: DataLike | None = None,
  492. ack_on_close: bool = False,
  493. ) -> threading.Event:
  494. """
  495. Send a Ping_.
  496. .. _Ping: https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.2
  497. A ping may serve as a keepalive or as a check that the remote endpoint
  498. received all messages up to this point
  499. Args:
  500. data: Payload of the ping. A :class:`str` will be encoded to UTF-8.
  501. If ``data`` is :obj:`None`, the payload is four random bytes.
  502. ack_on_close: when this option is :obj:`True`, the event will also
  503. be set when the connection is closed. While this avoids getting
  504. stuck waiting for a pong that will never arrive, it requires
  505. checking that the state of the connection is still ``OPEN`` to
  506. confirm that a pong was received, rather than the connection
  507. being closed.
  508. Returns:
  509. An event that will be set when the corresponding pong is received.
  510. You can ignore it if you don't intend to wait.
  511. ::
  512. pong_received = ws.ping()
  513. # only if you want to wait for the corresponding pong
  514. pong_received.wait()
  515. Raises:
  516. ConnectionClosed: When the connection is closed.
  517. ConcurrencyError: If another ping was sent with the same data and
  518. the corresponding pong wasn't received yet.
  519. """
  520. if isinstance(data, BytesLike):
  521. data = bytes(data)
  522. elif isinstance(data, str):
  523. data = data.encode()
  524. elif data is not None:
  525. raise TypeError("data must be str or bytes-like")
  526. with self.send_context():
  527. # Protect against duplicates if a payload is explicitly set.
  528. if data in self.pending_pings:
  529. raise ConcurrencyError("already waiting for a pong with the same data")
  530. # Generate a unique random payload otherwise.
  531. while data is None or data in self.pending_pings:
  532. data = struct.pack("!I", random.getrandbits(32))
  533. pong_received = threading.Event()
  534. ping_timestamp = time.monotonic()
  535. self.pending_pings[data] = (pong_received, ping_timestamp, ack_on_close)
  536. self.protocol.send_ping(data)
  537. return pong_received
  538. def pong(self, data: DataLike = b"") -> None:
  539. """
  540. Send a Pong_.
  541. .. _Pong: https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.3
  542. An unsolicited pong may serve as a unidirectional heartbeat.
  543. Args:
  544. data: Payload of the pong. A :class:`str` will be encoded to UTF-8.
  545. Raises:
  546. ConnectionClosed: When the connection is closed.
  547. """
  548. if isinstance(data, BytesLike):
  549. data = bytes(data)
  550. elif isinstance(data, str):
  551. data = data.encode()
  552. else:
  553. raise TypeError("data must be str or bytes-like")
  554. with self.send_context():
  555. self.protocol.send_pong(data)
  556. # Private methods
  557. def process_event(self, event: Event) -> None:
  558. """
  559. Process one incoming event.
  560. This method is overridden in subclasses to handle the handshake.
  561. """
  562. assert isinstance(event, Frame)
  563. if event.opcode in DATA_OPCODES:
  564. self.recv_messages.put(event)
  565. if event.opcode is Opcode.PONG:
  566. self.acknowledge_pings(bytes(event.data))
  567. def acknowledge_pings(self, data: bytes) -> None:
  568. """
  569. Acknowledge pings when receiving a pong.
  570. """
  571. with self.protocol_mutex:
  572. # Ignore unsolicited pong.
  573. if data not in self.pending_pings:
  574. return
  575. pong_timestamp = time.monotonic()
  576. # Sending a pong for only the most recent ping is legal.
  577. # Acknowledge all previous pings too in that case.
  578. ping_id = None
  579. ping_ids = []
  580. for ping_id, (
  581. pong_received,
  582. ping_timestamp,
  583. _ack_on_close,
  584. ) in self.pending_pings.items():
  585. ping_ids.append(ping_id)
  586. pong_received.set()
  587. if ping_id == data:
  588. self.latency = pong_timestamp - ping_timestamp
  589. break
  590. else:
  591. raise AssertionError("solicited pong not found in pings")
  592. # Remove acknowledged pings from self.pending_pings.
  593. for ping_id in ping_ids:
  594. del self.pending_pings[ping_id]
  595. def terminate_pending_pings(self) -> None:
  596. """
  597. Acknowledge pending pings when the connection is closed.
  598. """
  599. assert self.protocol_mutex.locked()
  600. assert self.protocol.state is CLOSED
  601. for pong_received, _ping_timestamp, ack_on_close in self.pending_pings.values():
  602. if ack_on_close:
  603. pong_received.set()
  604. self.pending_pings.clear()
  605. def keepalive(self) -> None:
  606. """
  607. Send a Ping frame and wait for a Pong frame at regular intervals.
  608. """
  609. assert self.ping_interval is not None
  610. try:
  611. while True:
  612. # If self.ping_timeout > self.latency > self.ping_interval,
  613. # pings will be sent immediately after receiving pongs.
  614. # The period will be longer than self.ping_interval.
  615. self.recv_events_thread.join(self.ping_interval - self.latency)
  616. if not self.recv_events_thread.is_alive():
  617. break
  618. try:
  619. pong_received = self.ping(ack_on_close=True)
  620. except ConnectionClosed:
  621. break
  622. if self.debug:
  623. self.logger.debug("% sent keepalive ping")
  624. if self.ping_timeout is not None:
  625. if pong_received.wait(self.ping_timeout):
  626. if self.debug:
  627. self.logger.debug("% received keepalive pong")
  628. else:
  629. if self.debug:
  630. self.logger.debug("- timed out waiting for keepalive pong")
  631. with self.send_context():
  632. self.protocol.fail(
  633. CloseCode.INTERNAL_ERROR,
  634. "keepalive ping timeout",
  635. )
  636. break
  637. except Exception:
  638. self.logger.error("keepalive ping failed", exc_info=True)
  639. def start_keepalive(self) -> None:
  640. """
  641. Run :meth:`keepalive` in a thread, unless keepalive is disabled.
  642. """
  643. if self.ping_interval is not None:
  644. # This thread is marked as daemon like self.recv_events_thread.
  645. self.keepalive_thread = threading.Thread(
  646. target=self.keepalive,
  647. daemon=True,
  648. )
  649. self.keepalive_thread.start()
  650. def recv_events(self) -> None:
  651. """
  652. Read incoming data from the socket and process events.
  653. Run this method in a thread as long as the connection is alive.
  654. ``recv_events()`` exits immediately when ``self.socket`` is closed.
  655. """
  656. try:
  657. while True:
  658. try:
  659. # If the assembler buffer is full, block until it drains.
  660. with self.recv_flow_control:
  661. pass
  662. if self.close_deadline is not None:
  663. self.socket.settimeout(self.close_deadline.timeout())
  664. data = self.socket.recv(self.recv_bufsize)
  665. except Exception as exc:
  666. if self.debug:
  667. self.logger.debug(
  668. "! error while receiving data",
  669. exc_info=True,
  670. )
  671. # When the closing handshake is initiated by our side,
  672. # recv() may block until send_context() closes the socket.
  673. # In that case, send_context() already set recv_exc.
  674. # Calling set_recv_exc() avoids overwriting it.
  675. with self.protocol_mutex:
  676. self.set_recv_exc(exc)
  677. break
  678. if data == b"":
  679. break
  680. # Acquire the connection lock.
  681. with self.protocol_mutex:
  682. # Feed incoming data to the protocol.
  683. self.protocol.receive_data(data)
  684. # This isn't expected to raise an exception.
  685. events = self.protocol.events_received()
  686. # Write outgoing data to the socket.
  687. try:
  688. self.send_data()
  689. except Exception as exc:
  690. if self.debug:
  691. self.logger.debug(
  692. "! error while sending data",
  693. exc_info=True,
  694. )
  695. # Similarly to the above, avoid overriding an exception
  696. # set by send_context(), in case of a race condition
  697. # i.e. send_context() closes the socket after recv()
  698. # returns above but before send_data() calls send().
  699. self.set_recv_exc(exc)
  700. break
  701. # If needed, set the close deadline based on the close timeout.
  702. if self.protocol.close_expected():
  703. if self.close_deadline is None:
  704. self.close_deadline = Deadline(self.close_timeout)
  705. # Unlock conn_mutex before processing events. Else, the
  706. # application can't send messages in response to events.
  707. # If self.send_data raised an exception, then events are lost.
  708. # Given that automatic responses write small amounts of data,
  709. # this should be uncommon, so we don't handle the edge case.
  710. for event in events:
  711. # This isn't expected to raise an exception.
  712. self.process_event(event)
  713. # Breaking out of the while True: ... loop means that we believe
  714. # that the socket doesn't work anymore.
  715. with self.protocol_mutex:
  716. # Feed the end of the data stream to the protocol.
  717. self.protocol.receive_eof()
  718. # This isn't expected to raise an exception.
  719. events = self.protocol.events_received()
  720. # There is no error handling because send_data() can only write
  721. # the end of the data stream and it handles errors by itself.
  722. self.send_data()
  723. # This code path is triggered when receiving an HTTP response
  724. # without a Content-Length header. This is the only case where
  725. # reading until EOF generates an event; all other events have
  726. # a known length. Ignore for coverage measurement because tests
  727. # are in test_client.py rather than test_connection.py.
  728. for event in events: # pragma: no cover
  729. # This isn't expected to raise an exception.
  730. self.process_event(event)
  731. except Exception as exc:
  732. # This branch should never run. It's a safety net in case of bugs.
  733. self.logger.error("unexpected internal error", exc_info=True)
  734. with self.protocol_mutex:
  735. self.set_recv_exc(exc)
  736. finally:
  737. # This isn't expected to raise an exception.
  738. self.close_socket()
  739. @contextlib.contextmanager
  740. def send_context(
  741. self,
  742. *,
  743. expected_state: State = OPEN, # CONNECTING during the opening handshake
  744. ) -> Iterator[None]:
  745. """
  746. Create a context for writing to the connection from user code.
  747. On entry, :meth:`send_context` acquires the connection lock and checks
  748. that the connection is open; on exit, it writes outgoing data to the
  749. socket and releases the connection lock::
  750. with self.send_context():
  751. self.protocol.send_text(message.encode())
  752. When the connection isn't open on entry, when the connection is expected
  753. to close on exit, or when an unexpected error happens, terminating the
  754. connection, :meth:`send_context` waits until the connection is closed
  755. then raises :exc:`~websockets.exceptions.ConnectionClosed`.
  756. """
  757. # Should we wait until the connection is closed?
  758. wait_for_close = False
  759. # Should we close the socket and raise ConnectionClosed?
  760. raise_close_exc = False
  761. # What exception should we chain ConnectionClosed to?
  762. original_exc: BaseException | None = None
  763. # Acquire the protocol lock.
  764. with self.protocol_mutex:
  765. if self.protocol.state is expected_state:
  766. # Let the caller interact with the protocol.
  767. try:
  768. yield
  769. except (ProtocolError, ConcurrencyError):
  770. # The protocol state wasn't changed. Exit immediately.
  771. raise
  772. except Exception as exc:
  773. self.logger.error("unexpected internal error", exc_info=True)
  774. # This branch should never run. It's a safety net in case of
  775. # bugs. Since we don't know what happened, we will close the
  776. # connection and raise the exception to the caller.
  777. wait_for_close = False
  778. raise_close_exc = True
  779. original_exc = exc
  780. else:
  781. # Check if the connection is expected to close soon.
  782. if self.protocol.close_expected():
  783. wait_for_close = True
  784. # Set the close deadline based on the close timeout.
  785. # Since we tested earlier that protocol.state is OPEN
  786. # (or CONNECTING) and we didn't release protocol_mutex,
  787. # self.close_deadline is still None.
  788. assert self.close_deadline is None
  789. self.close_deadline = Deadline(self.close_timeout)
  790. # Write outgoing data to the socket.
  791. try:
  792. self.send_data()
  793. except Exception as exc:
  794. if self.debug:
  795. self.logger.debug(
  796. "! error while sending data",
  797. exc_info=True,
  798. )
  799. # While the only expected exception here is OSError,
  800. # other exceptions would be treated identically.
  801. wait_for_close = False
  802. raise_close_exc = True
  803. original_exc = exc
  804. else: # self.protocol.state is not expected_state
  805. # Minor layering violation: we assume that the connection
  806. # will be closing soon if it isn't in the expected state.
  807. wait_for_close = True
  808. # Calculate close_deadline if it wasn't set yet.
  809. if self.close_deadline is None:
  810. self.close_deadline = Deadline(self.close_timeout)
  811. raise_close_exc = True
  812. # To avoid a deadlock, release the connection lock by exiting the
  813. # context manager before waiting for recv_events() to terminate.
  814. # If the connection is expected to close soon and the close timeout
  815. # elapses, close the socket to terminate the connection.
  816. if wait_for_close:
  817. # Thread.join() returns immediately if timeout is negative.
  818. assert self.close_deadline is not None
  819. timeout = self.close_deadline.timeout(raise_if_elapsed=False)
  820. self.recv_events_thread.join(timeout)
  821. if self.recv_events_thread.is_alive():
  822. # There's no risk of overwriting another error because
  823. # original_exc is never set when wait_for_close is True.
  824. assert original_exc is None
  825. original_exc = TimeoutError("timed out while closing connection")
  826. # Set recv_exc before closing the socket in order to get
  827. # proper exception reporting.
  828. raise_close_exc = True
  829. with self.protocol_mutex:
  830. self.set_recv_exc(original_exc)
  831. # If an error occurred, close the socket to terminate the connection and
  832. # raise an exception.
  833. if raise_close_exc:
  834. self.close_socket()
  835. # Wait for the protocol state to be CLOSED before accessing close_exc.
  836. self.recv_events_thread.join()
  837. raise self.protocol.close_exc from original_exc
  838. def send_data(self) -> None:
  839. """
  840. Send outgoing data.
  841. This method requires holding protocol_mutex.
  842. """
  843. assert self.protocol_mutex.locked()
  844. for data in self.protocol.data_to_send():
  845. if data:
  846. if self.close_deadline is not None:
  847. self.socket.settimeout(self.close_deadline.timeout())
  848. self.socket.sendall(data)
  849. else:
  850. try:
  851. self.socket.shutdown(socket.SHUT_WR)
  852. except OSError: # socket already closed
  853. pass
  854. def set_recv_exc(self, exc: BaseException | None) -> None:
  855. """
  856. Set recv_exc, if not set yet.
  857. This method requires holding protocol_mutex and must be called only from
  858. the thread running recv_events().
  859. """
  860. assert self.protocol_mutex.locked()
  861. if self.recv_exc is None:
  862. self.recv_exc = exc
  863. def close_socket(self) -> None:
  864. """
  865. Shutdown and close socket. Close message assembler.
  866. Calling close_socket() guarantees that recv_events() terminates. Indeed,
  867. recv_events() may block only on socket.recv() or on recv_messages.put().
  868. """
  869. # shutdown() is required to interrupt recv() on Linux.
  870. try:
  871. self.socket.shutdown(socket.SHUT_RDWR)
  872. except OSError: # socket already closed
  873. pass
  874. self.socket.close()
  875. # Calling protocol.receive_eof() is safe because it's idempotent.
  876. # This guarantees that the protocol state becomes CLOSED.
  877. with self.protocol_mutex:
  878. self.protocol.receive_eof()
  879. assert self.protocol.state is CLOSED
  880. # Abort recv() with a ConnectionClosed exception.
  881. self.recv_messages.close()
  882. # Acknowledge pings sent with the ack_on_close option.
  883. self.terminate_pending_pings()