protocol.py 50 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361
  1. """
  2. :mod:`websockets.legacy.protocol` handles WebSocket control and data frames.
  3. See `sections 4 to 8 of RFC 6455`_.
  4. .. _sections 4 to 8 of RFC 6455: http://tools.ietf.org/html/rfc6455#section-4
  5. """
  6. import asyncio
  7. import codecs
  8. import collections
  9. import enum
  10. import logging
  11. import random
  12. import struct
  13. import sys
  14. import warnings
  15. from typing import (
  16. Any,
  17. AsyncIterable,
  18. AsyncIterator,
  19. Awaitable,
  20. Deque,
  21. Dict,
  22. Iterable,
  23. List,
  24. Mapping,
  25. Optional,
  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.base import Extension
  39. from ..frames import (
  40. OP_BINARY,
  41. OP_CLOSE,
  42. OP_CONT,
  43. OP_PING,
  44. OP_PONG,
  45. OP_TEXT,
  46. Opcode,
  47. parse_close,
  48. prepare_ctrl,
  49. prepare_data,
  50. serialize_close,
  51. )
  52. from ..typing import Data, Subprotocol
  53. from .framing import Frame
  54. __all__ = ["WebSocketCommonProtocol"]
  55. logger = logging.getLogger("websockets.protocol")
  56. # A WebSocket connection goes through the following four states, in order:
  57. class State(enum.IntEnum):
  58. CONNECTING, OPEN, CLOSING, CLOSED = range(4)
  59. # In order to ensure consistency, the code always checks the current value of
  60. # WebSocketCommonProtocol.state before assigning a new value and never yields
  61. # between the check and the assignment.
  62. class WebSocketCommonProtocol(asyncio.Protocol):
  63. """
  64. :class:`~asyncio.Protocol` subclass implementing the data transfer phase.
  65. Once the WebSocket connection is established, during the data transfer
  66. phase, the protocol is almost symmetrical between the server side and the
  67. client side. :class:`WebSocketCommonProtocol` implements logic that's
  68. shared between servers and clients.
  69. Subclasses such as
  70. :class:`~websockets.legacy.server.WebSocketServerProtocol` and
  71. :class:`~websockets.legacy.client.WebSocketClientProtocol` implement the
  72. opening handshake, which is different between servers and clients.
  73. """
  74. # There are only two differences between the client-side and server-side
  75. # behavior: masking the payload and closing the underlying TCP connection.
  76. # Set is_client = True/False and side = "client"/"server" to pick a side.
  77. is_client: bool
  78. side: str = "undefined"
  79. def __init__(
  80. self,
  81. *,
  82. ping_interval: Optional[float] = 20,
  83. ping_timeout: Optional[float] = 20,
  84. close_timeout: Optional[float] = None,
  85. max_size: Optional[int] = 2 ** 20,
  86. max_queue: Optional[int] = 2 ** 5,
  87. read_limit: int = 2 ** 16,
  88. write_limit: int = 2 ** 16,
  89. loop: Optional[asyncio.AbstractEventLoop] = None,
  90. # The following arguments are kept only for backwards compatibility.
  91. host: Optional[str] = None,
  92. port: Optional[int] = None,
  93. secure: Optional[bool] = None,
  94. legacy_recv: bool = False,
  95. timeout: Optional[float] = None,
  96. ) -> None:
  97. # Backwards compatibility: close_timeout used to be called timeout.
  98. if timeout is None:
  99. timeout = 10
  100. else:
  101. warnings.warn("rename timeout to close_timeout", DeprecationWarning)
  102. # If both are specified, timeout is ignored.
  103. if close_timeout is None:
  104. close_timeout = timeout
  105. self.ping_interval = ping_interval
  106. self.ping_timeout = ping_timeout
  107. self.close_timeout = close_timeout
  108. self.max_size = max_size
  109. self.max_queue = max_queue
  110. self.read_limit = read_limit
  111. self.write_limit = write_limit
  112. if loop is None:
  113. loop = asyncio.get_event_loop()
  114. self.loop = loop
  115. self._host = host
  116. self._port = port
  117. self._secure = secure
  118. self.legacy_recv = legacy_recv
  119. # Configure read buffer limits. The high-water limit is defined by
  120. # ``self.read_limit``. The ``limit`` argument controls the line length
  121. # limit and half the buffer limit of :class:`~asyncio.StreamReader`.
  122. # That's why it must be set to half of ``self.read_limit``.
  123. self.reader = asyncio.StreamReader(limit=read_limit // 2, loop=loop)
  124. # Copied from asyncio.FlowControlMixin
  125. self._paused = False
  126. self._drain_waiter: Optional[asyncio.Future[None]] = None
  127. self._drain_lock = asyncio.Lock(
  128. loop=loop if sys.version_info[:2] < (3, 8) else None
  129. )
  130. # This class implements the data transfer and closing handshake, which
  131. # are shared between the client-side and the server-side.
  132. # Subclasses implement the opening handshake and, on success, execute
  133. # :meth:`connection_open` to change the state to OPEN.
  134. self.state = State.CONNECTING
  135. logger.debug("%s - state = CONNECTING", self.side)
  136. # HTTP protocol parameters.
  137. self.path: str
  138. self.request_headers: Headers
  139. self.response_headers: Headers
  140. # WebSocket protocol parameters.
  141. self.extensions: List[Extension] = []
  142. self.subprotocol: Optional[Subprotocol] = None
  143. # The close code and reason are set when receiving a close frame or
  144. # losing the TCP connection.
  145. self.close_code: int
  146. self.close_reason: str
  147. # Completed when the connection state becomes CLOSED. Translates the
  148. # :meth:`connection_lost` callback to a :class:`~asyncio.Future`
  149. # that can be awaited. (Other :class:`~asyncio.Protocol` callbacks are
  150. # translated by ``self.stream_reader``).
  151. self.connection_lost_waiter: asyncio.Future[None] = loop.create_future()
  152. # Queue of received messages.
  153. self.messages: Deque[Data] = collections.deque()
  154. self._pop_message_waiter: Optional[asyncio.Future[None]] = None
  155. self._put_message_waiter: Optional[asyncio.Future[None]] = None
  156. # Protect sending fragmented messages.
  157. self._fragmented_message_waiter: Optional[asyncio.Future[None]] = None
  158. # Mapping of ping IDs to pong waiters, in chronological order.
  159. self.pings: Dict[bytes, asyncio.Future[None]] = {}
  160. # Task running the data transfer.
  161. self.transfer_data_task: asyncio.Task[None]
  162. # Exception that occurred during data transfer, if any.
  163. self.transfer_data_exc: Optional[BaseException] = None
  164. # Task sending keepalive pings.
  165. self.keepalive_ping_task: asyncio.Task[None]
  166. # Task closing the TCP connection.
  167. self.close_connection_task: asyncio.Task[None]
  168. # Copied from asyncio.FlowControlMixin
  169. async def _drain_helper(self) -> None: # pragma: no cover
  170. if self.connection_lost_waiter.done():
  171. raise ConnectionResetError("Connection lost")
  172. if not self._paused:
  173. return
  174. waiter = self._drain_waiter
  175. assert waiter is None or waiter.cancelled()
  176. waiter = self.loop.create_future()
  177. self._drain_waiter = waiter
  178. await waiter
  179. # Copied from asyncio.StreamWriter
  180. async def _drain(self) -> None: # pragma: no cover
  181. if self.reader is not None:
  182. exc = self.reader.exception()
  183. if exc is not None:
  184. raise exc
  185. if self.transport is not None:
  186. if self.transport.is_closing():
  187. # Yield to the event loop so connection_lost() may be
  188. # called. Without this, _drain_helper() would return
  189. # immediately, and code that calls
  190. # write(...); yield from drain()
  191. # in a loop would never call connection_lost(), so it
  192. # would not see an error when the socket is closed.
  193. await asyncio.sleep(
  194. 0, loop=self.loop if sys.version_info[:2] < (3, 8) else None
  195. )
  196. await self._drain_helper()
  197. def connection_open(self) -> None:
  198. """
  199. Callback when the WebSocket opening handshake completes.
  200. Enter the OPEN state and start the data transfer phase.
  201. """
  202. # 4.1. The WebSocket Connection is Established.
  203. assert self.state is State.CONNECTING
  204. self.state = State.OPEN
  205. logger.debug("%s - state = OPEN", self.side)
  206. # Start the task that receives incoming WebSocket messages.
  207. self.transfer_data_task = self.loop.create_task(self.transfer_data())
  208. # Start the task that sends pings at regular intervals.
  209. self.keepalive_ping_task = self.loop.create_task(self.keepalive_ping())
  210. # Start the task that eventually closes the TCP connection.
  211. self.close_connection_task = self.loop.create_task(self.close_connection())
  212. @property
  213. def host(self) -> Optional[str]:
  214. alternative = "remote_address" if self.is_client else "local_address"
  215. warnings.warn(f"use {alternative}[0] instead of host", DeprecationWarning)
  216. return self._host
  217. @property
  218. def port(self) -> Optional[int]:
  219. alternative = "remote_address" if self.is_client else "local_address"
  220. warnings.warn(f"use {alternative}[1] instead of port", DeprecationWarning)
  221. return self._port
  222. @property
  223. def secure(self) -> Optional[bool]:
  224. warnings.warn("don't use secure", DeprecationWarning)
  225. return self._secure
  226. # Public API
  227. @property
  228. def local_address(self) -> Any:
  229. """
  230. Local address of the connection as a ``(host, port)`` tuple.
  231. When the connection isn't open, ``local_address`` is ``None``.
  232. """
  233. try:
  234. transport = self.transport
  235. except AttributeError:
  236. return None
  237. else:
  238. return transport.get_extra_info("sockname")
  239. @property
  240. def remote_address(self) -> Any:
  241. """
  242. Remote address of the connection as a ``(host, port)`` tuple.
  243. When the connection isn't open, ``remote_address`` is ``None``.
  244. """
  245. try:
  246. transport = self.transport
  247. except AttributeError:
  248. return None
  249. else:
  250. return transport.get_extra_info("peername")
  251. @property
  252. def open(self) -> bool:
  253. """
  254. ``True`` when the connection is usable.
  255. It may be used to detect disconnections. However, this approach is
  256. discouraged per the EAFP_ principle.
  257. When ``open`` is ``False``, using the connection raises a
  258. :exc:`~websockets.exceptions.ConnectionClosed` exception.
  259. .. _EAFP: https://docs.python.org/3/glossary.html#term-eafp
  260. """
  261. return self.state is State.OPEN and not self.transfer_data_task.done()
  262. @property
  263. def closed(self) -> bool:
  264. """
  265. ``True`` once the connection is closed.
  266. Be aware that both :attr:`open` and :attr:`closed` are ``False`` during
  267. the opening and closing sequences.
  268. """
  269. return self.state is State.CLOSED
  270. async def wait_closed(self) -> None:
  271. """
  272. Wait until the connection is closed.
  273. This is identical to :attr:`closed`, except it can be awaited.
  274. This can make it easier to handle connection termination, regardless
  275. of its cause, in tasks that interact with the WebSocket connection.
  276. """
  277. await asyncio.shield(self.connection_lost_waiter)
  278. async def __aiter__(self) -> AsyncIterator[Data]:
  279. """
  280. Iterate on received messages.
  281. Exit normally when the connection is closed with code 1000 or 1001.
  282. Raise an exception in other cases.
  283. """
  284. try:
  285. while True:
  286. yield await self.recv()
  287. except ConnectionClosedOK:
  288. return
  289. async def recv(self) -> Data:
  290. """
  291. Receive the next message.
  292. Return a :class:`str` for a text frame and :class:`bytes` for a binary
  293. frame.
  294. When the end of the message stream is reached, :meth:`recv` raises
  295. :exc:`~websockets.exceptions.ConnectionClosed`. Specifically, it
  296. raises :exc:`~websockets.exceptions.ConnectionClosedOK` after a normal
  297. connection closure and
  298. :exc:`~websockets.exceptions.ConnectionClosedError` after a protocol
  299. error or a network failure.
  300. Canceling :meth:`recv` is safe. There's no risk of losing the next
  301. message. The next invocation of :meth:`recv` will return it. This
  302. makes it possible to enforce a timeout by wrapping :meth:`recv` in
  303. :func:`~asyncio.wait_for`.
  304. :raises ~websockets.exceptions.ConnectionClosed: when the
  305. connection is closed
  306. :raises RuntimeError: if two coroutines call :meth:`recv` concurrently
  307. """
  308. if self._pop_message_waiter is not None:
  309. raise RuntimeError(
  310. "cannot call recv while another coroutine "
  311. "is already waiting for the next message"
  312. )
  313. # Don't await self.ensure_open() here:
  314. # - messages could be available in the queue even if the connection
  315. # is closed;
  316. # - messages could be received before the closing frame even if the
  317. # connection is closing.
  318. # Wait until there's a message in the queue (if necessary) or the
  319. # connection is closed.
  320. while len(self.messages) <= 0:
  321. pop_message_waiter: asyncio.Future[None] = self.loop.create_future()
  322. self._pop_message_waiter = pop_message_waiter
  323. try:
  324. # If asyncio.wait() is canceled, it doesn't cancel
  325. # pop_message_waiter and self.transfer_data_task.
  326. await asyncio.wait(
  327. [pop_message_waiter, self.transfer_data_task],
  328. loop=self.loop if sys.version_info[:2] < (3, 8) else None,
  329. return_when=asyncio.FIRST_COMPLETED,
  330. )
  331. finally:
  332. self._pop_message_waiter = None
  333. # If asyncio.wait(...) exited because self.transfer_data_task
  334. # completed before receiving a new message, raise a suitable
  335. # exception (or return None if legacy_recv is enabled).
  336. if not pop_message_waiter.done():
  337. if self.legacy_recv:
  338. return None # type: ignore
  339. else:
  340. # Wait until the connection is closed to raise
  341. # ConnectionClosed with the correct code and reason.
  342. await self.ensure_open()
  343. # Pop a message from the queue.
  344. message = self.messages.popleft()
  345. # Notify transfer_data().
  346. if self._put_message_waiter is not None:
  347. self._put_message_waiter.set_result(None)
  348. self._put_message_waiter = None
  349. return message
  350. async def send(
  351. self, message: Union[Data, Iterable[Data], AsyncIterable[Data]]
  352. ) -> None:
  353. """
  354. Send a message.
  355. A string (:class:`str`) is sent as a `Text frame`_. A bytestring or
  356. bytes-like object (:class:`bytes`, :class:`bytearray`, or
  357. :class:`memoryview`) is sent as a `Binary frame`_.
  358. .. _Text frame: https://tools.ietf.org/html/rfc6455#section-5.6
  359. .. _Binary frame: https://tools.ietf.org/html/rfc6455#section-5.6
  360. :meth:`send` also accepts an iterable or an asynchronous iterable of
  361. strings, bytestrings, or bytes-like objects. In that case the message
  362. is fragmented. Each item is treated as a message fragment and sent in
  363. its own frame. All items must be of the same type, or else
  364. :meth:`send` will raise a :exc:`TypeError` and the connection will be
  365. closed.
  366. :meth:`send` rejects dict-like objects because this is often an error.
  367. If you wish to send the keys of a dict-like object as fragments, call
  368. its :meth:`~dict.keys` method and pass the result to :meth:`send`.
  369. Canceling :meth:`send` is discouraged. Instead, you should close the
  370. connection with :meth:`close`. Indeed, there are only two situations
  371. where :meth:`send` may yield control to the event loop:
  372. 1. The write buffer is full. If you don't want to wait until enough
  373. data is sent, your only alternative is to close the connection.
  374. :meth:`close` will likely time out then abort the TCP connection.
  375. 2. ``message`` is an asynchronous iterator that yields control.
  376. Stopping in the middle of a fragmented message will cause a
  377. protocol error. Closing the connection has the same effect.
  378. :raises TypeError: for unsupported inputs
  379. """
  380. await self.ensure_open()
  381. # While sending a fragmented message, prevent sending other messages
  382. # until all fragments are sent.
  383. while self._fragmented_message_waiter is not None:
  384. await asyncio.shield(self._fragmented_message_waiter)
  385. # Unfragmented message -- this case must be handled first because
  386. # strings and bytes-like objects are iterable.
  387. if isinstance(message, (str, bytes, bytearray, memoryview)):
  388. opcode, data = prepare_data(message)
  389. await self.write_frame(True, opcode, data)
  390. # Catch a common mistake -- passing a dict to send().
  391. elif isinstance(message, Mapping):
  392. raise TypeError("data is a dict-like object")
  393. # Fragmented message -- regular iterator.
  394. elif isinstance(message, Iterable):
  395. # Work around https://github.com/python/mypy/issues/6227
  396. message = cast(Iterable[Data], message)
  397. iter_message = iter(message)
  398. try:
  399. message_chunk = next(iter_message)
  400. except StopIteration:
  401. return
  402. opcode, data = prepare_data(message_chunk)
  403. self._fragmented_message_waiter = asyncio.Future()
  404. try:
  405. # First fragment.
  406. await self.write_frame(False, opcode, data)
  407. # Other fragments.
  408. for message_chunk in iter_message:
  409. confirm_opcode, data = prepare_data(message_chunk)
  410. if confirm_opcode != opcode:
  411. raise TypeError("data contains inconsistent types")
  412. await self.write_frame(False, OP_CONT, data)
  413. # Final fragment.
  414. await self.write_frame(True, OP_CONT, b"")
  415. except Exception:
  416. # We're half-way through a fragmented message and we can't
  417. # complete it. This makes the connection unusable.
  418. self.fail_connection(1011)
  419. raise
  420. finally:
  421. self._fragmented_message_waiter.set_result(None)
  422. self._fragmented_message_waiter = None
  423. # Fragmented message -- asynchronous iterator
  424. elif isinstance(message, AsyncIterable):
  425. # aiter_message = aiter(message) without aiter
  426. # https://github.com/python/mypy/issues/5738
  427. aiter_message = type(message).__aiter__(message) # type: ignore
  428. try:
  429. # message_chunk = anext(aiter_message) without anext
  430. # https://github.com/python/mypy/issues/5738
  431. message_chunk = await type(aiter_message).__anext__( # type: ignore
  432. aiter_message
  433. )
  434. except StopAsyncIteration:
  435. return
  436. opcode, data = prepare_data(message_chunk)
  437. self._fragmented_message_waiter = asyncio.Future()
  438. try:
  439. # First fragment.
  440. await self.write_frame(False, opcode, data)
  441. # Other fragments.
  442. # https://github.com/python/mypy/issues/5738
  443. # coverage reports this code as not covered, but it is
  444. # exercised by tests - changing it breaks the tests!
  445. async for message_chunk in aiter_message: # type: ignore # pragma: no cover # noqa
  446. confirm_opcode, data = prepare_data(message_chunk)
  447. if confirm_opcode != opcode:
  448. raise TypeError("data contains inconsistent types")
  449. await self.write_frame(False, OP_CONT, data)
  450. # Final fragment.
  451. await self.write_frame(True, OP_CONT, b"")
  452. except Exception:
  453. # We're half-way through a fragmented message and we can't
  454. # complete it. This makes the connection unusable.
  455. self.fail_connection(1011)
  456. raise
  457. finally:
  458. self._fragmented_message_waiter.set_result(None)
  459. self._fragmented_message_waiter = None
  460. else:
  461. raise TypeError("data must be bytes, str, or iterable")
  462. async def close(self, code: int = 1000, reason: str = "") -> None:
  463. """
  464. Perform the closing handshake.
  465. :meth:`close` waits for the other end to complete the handshake and
  466. for the TCP connection to terminate. As a consequence, there's no need
  467. to await :meth:`wait_closed`; :meth:`close` already does it.
  468. :meth:`close` is idempotent: it doesn't do anything once the
  469. connection is closed.
  470. Wrapping :func:`close` in :func:`~asyncio.create_task` is safe, given
  471. that errors during connection termination aren't particularly useful.
  472. Canceling :meth:`close` is discouraged. If it takes too long, you can
  473. set a shorter ``close_timeout``. If you don't want to wait, let the
  474. Python process exit, then the OS will close the TCP connection.
  475. :param code: WebSocket close code
  476. :param reason: WebSocket close reason
  477. """
  478. try:
  479. await asyncio.wait_for(
  480. self.write_close_frame(serialize_close(code, reason)),
  481. self.close_timeout,
  482. loop=self.loop if sys.version_info[:2] < (3, 8) else None,
  483. )
  484. except asyncio.TimeoutError:
  485. # If the close frame cannot be sent because the send buffers
  486. # are full, the closing handshake won't complete anyway.
  487. # Fail the connection to shut down faster.
  488. self.fail_connection()
  489. # If no close frame is received within the timeout, wait_for() cancels
  490. # the data transfer task and raises TimeoutError.
  491. # If close() is called multiple times concurrently and one of these
  492. # calls hits the timeout, the data transfer task will be cancelled.
  493. # Other calls will receive a CancelledError here.
  494. try:
  495. # If close() is canceled during the wait, self.transfer_data_task
  496. # is canceled before the timeout elapses.
  497. await asyncio.wait_for(
  498. self.transfer_data_task,
  499. self.close_timeout,
  500. loop=self.loop if sys.version_info[:2] < (3, 8) else None,
  501. )
  502. except (asyncio.TimeoutError, asyncio.CancelledError):
  503. pass
  504. # Wait for the close connection task to close the TCP connection.
  505. await asyncio.shield(self.close_connection_task)
  506. async def ping(self, data: Optional[Data] = None) -> Awaitable[None]:
  507. """
  508. Send a ping.
  509. Return a :class:`~asyncio.Future` that will be completed when the
  510. corresponding pong is received. You can ignore it if you don't intend
  511. to wait.
  512. A ping may serve as a keepalive or as a check that the remote endpoint
  513. received all messages up to this point::
  514. pong_waiter = await ws.ping()
  515. await pong_waiter # only if you want to wait for the pong
  516. By default, the ping contains four random bytes. This payload may be
  517. overridden with the optional ``data`` argument which must be a string
  518. (which will be encoded to UTF-8) or a bytes-like object.
  519. Canceling :meth:`ping` is discouraged. If :meth:`ping` doesn't return
  520. immediately, it means the write buffer is full. If you don't want to
  521. wait, you should close the connection.
  522. Canceling the :class:`~asyncio.Future` returned by :meth:`ping` has no
  523. effect.
  524. """
  525. await self.ensure_open()
  526. if data is not None:
  527. data = prepare_ctrl(data)
  528. # Protect against duplicates if a payload is explicitly set.
  529. if data in self.pings:
  530. raise ValueError("already waiting for a pong with the same data")
  531. # Generate a unique random payload otherwise.
  532. while data is None or data in self.pings:
  533. data = struct.pack("!I", random.getrandbits(32))
  534. self.pings[data] = self.loop.create_future()
  535. await self.write_frame(True, OP_PING, data)
  536. return asyncio.shield(self.pings[data])
  537. async def pong(self, data: Data = b"") -> None:
  538. """
  539. Send a pong.
  540. An unsolicited pong may serve as a unidirectional heartbeat.
  541. The payload may be set with the optional ``data`` argument which must
  542. be a string (which will be encoded to UTF-8) or a bytes-like object.
  543. Canceling :meth:`pong` is discouraged for the same reason as
  544. :meth:`ping`.
  545. """
  546. await self.ensure_open()
  547. data = prepare_ctrl(data)
  548. await self.write_frame(True, OP_PONG, data)
  549. # Private methods - no guarantees.
  550. def connection_closed_exc(self) -> ConnectionClosed:
  551. exception: ConnectionClosed
  552. if self.close_code == 1000 or self.close_code == 1001:
  553. exception = ConnectionClosedOK(self.close_code, self.close_reason)
  554. else:
  555. exception = ConnectionClosedError(self.close_code, self.close_reason)
  556. # Chain to the exception that terminated data transfer, if any.
  557. exception.__cause__ = self.transfer_data_exc
  558. return exception
  559. async def ensure_open(self) -> None:
  560. """
  561. Check that the WebSocket connection is open.
  562. Raise :exc:`~websockets.exceptions.ConnectionClosed` if it isn't.
  563. """
  564. # Handle cases from most common to least common for performance.
  565. if self.state is State.OPEN:
  566. # If self.transfer_data_task exited without a closing handshake,
  567. # self.close_connection_task may be closing the connection, going
  568. # straight from OPEN to CLOSED.
  569. if self.transfer_data_task.done():
  570. await asyncio.shield(self.close_connection_task)
  571. raise self.connection_closed_exc()
  572. else:
  573. return
  574. if self.state is State.CLOSED:
  575. raise self.connection_closed_exc()
  576. if self.state is State.CLOSING:
  577. # If we started the closing handshake, wait for its completion to
  578. # get the proper close code and reason. self.close_connection_task
  579. # will complete within 4 or 5 * close_timeout after close(). The
  580. # CLOSING state also occurs when failing the connection. In that
  581. # case self.close_connection_task will complete even faster.
  582. await asyncio.shield(self.close_connection_task)
  583. raise self.connection_closed_exc()
  584. # Control may only reach this point in buggy third-party subclasses.
  585. assert self.state is State.CONNECTING
  586. raise InvalidState("WebSocket connection isn't established yet")
  587. async def transfer_data(self) -> None:
  588. """
  589. Read incoming messages and put them in a queue.
  590. This coroutine runs in a task until the closing handshake is started.
  591. """
  592. try:
  593. while True:
  594. message = await self.read_message()
  595. # Exit the loop when receiving a close frame.
  596. if message is None:
  597. break
  598. # Wait until there's room in the queue (if necessary).
  599. if self.max_queue is not None:
  600. while len(self.messages) >= self.max_queue:
  601. self._put_message_waiter = self.loop.create_future()
  602. try:
  603. await asyncio.shield(self._put_message_waiter)
  604. finally:
  605. self._put_message_waiter = None
  606. # Put the message in the queue.
  607. self.messages.append(message)
  608. # Notify recv().
  609. if self._pop_message_waiter is not None:
  610. self._pop_message_waiter.set_result(None)
  611. self._pop_message_waiter = None
  612. except asyncio.CancelledError as exc:
  613. self.transfer_data_exc = exc
  614. # If fail_connection() cancels this task, avoid logging the error
  615. # twice and failing the connection again.
  616. raise
  617. except ProtocolError as exc:
  618. self.transfer_data_exc = exc
  619. self.fail_connection(1002)
  620. except (ConnectionError, TimeoutError, EOFError) as exc:
  621. # Reading data with self.reader.readexactly may raise:
  622. # - most subclasses of ConnectionError if the TCP connection
  623. # breaks, is reset, or is aborted;
  624. # - TimeoutError if the TCP connection times out;
  625. # - IncompleteReadError, a subclass of EOFError, if fewer
  626. # bytes are available than requested.
  627. self.transfer_data_exc = exc
  628. self.fail_connection(1006)
  629. except UnicodeDecodeError as exc:
  630. self.transfer_data_exc = exc
  631. self.fail_connection(1007)
  632. except PayloadTooBig as exc:
  633. self.transfer_data_exc = exc
  634. self.fail_connection(1009)
  635. except Exception as exc:
  636. # This shouldn't happen often because exceptions expected under
  637. # regular circumstances are handled above. If it does, consider
  638. # catching and handling more exceptions.
  639. logger.error("Error in data transfer", exc_info=True)
  640. self.transfer_data_exc = exc
  641. self.fail_connection(1011)
  642. async def read_message(self) -> Optional[Data]:
  643. """
  644. Read a single message from the connection.
  645. Re-assemble data frames if the message is fragmented.
  646. Return ``None`` when the closing handshake is started.
  647. """
  648. frame = await self.read_data_frame(max_size=self.max_size)
  649. # A close frame was received.
  650. if frame is None:
  651. return None
  652. if frame.opcode == OP_TEXT:
  653. text = True
  654. elif frame.opcode == OP_BINARY:
  655. text = False
  656. else: # frame.opcode == OP_CONT
  657. raise ProtocolError("unexpected opcode")
  658. # Shortcut for the common case - no fragmentation
  659. if frame.fin:
  660. return frame.data.decode("utf-8") if text else frame.data
  661. # 5.4. Fragmentation
  662. chunks: List[Data] = []
  663. max_size = self.max_size
  664. if text:
  665. decoder_factory = codecs.getincrementaldecoder("utf-8")
  666. decoder = decoder_factory(errors="strict")
  667. if max_size is None:
  668. def append(frame: Frame) -> None:
  669. nonlocal chunks
  670. chunks.append(decoder.decode(frame.data, frame.fin))
  671. else:
  672. def append(frame: Frame) -> None:
  673. nonlocal chunks, max_size
  674. chunks.append(decoder.decode(frame.data, frame.fin))
  675. assert isinstance(max_size, int)
  676. max_size -= len(frame.data)
  677. else:
  678. if max_size is None:
  679. def append(frame: Frame) -> None:
  680. nonlocal chunks
  681. chunks.append(frame.data)
  682. else:
  683. def append(frame: Frame) -> None:
  684. nonlocal chunks, max_size
  685. chunks.append(frame.data)
  686. assert isinstance(max_size, int)
  687. max_size -= len(frame.data)
  688. append(frame)
  689. while not frame.fin:
  690. frame = await self.read_data_frame(max_size=max_size)
  691. if frame is None:
  692. raise ProtocolError("incomplete fragmented message")
  693. if frame.opcode != OP_CONT:
  694. raise ProtocolError("unexpected opcode")
  695. append(frame)
  696. # mypy cannot figure out that chunks have the proper type.
  697. return ("" if text else b"").join(chunks) # type: ignore
  698. async def read_data_frame(self, max_size: Optional[int]) -> Optional[Frame]:
  699. """
  700. Read a single data frame from the connection.
  701. Process control frames received before the next data frame.
  702. Return ``None`` if a close frame is encountered before any data frame.
  703. """
  704. # 6.2. Receiving Data
  705. while True:
  706. frame = await self.read_frame(max_size)
  707. # 5.5. Control Frames
  708. if frame.opcode == OP_CLOSE:
  709. # 7.1.5. The WebSocket Connection Close Code
  710. # 7.1.6. The WebSocket Connection Close Reason
  711. self.close_code, self.close_reason = parse_close(frame.data)
  712. try:
  713. # Echo the original data instead of re-serializing it with
  714. # serialize_close() because that fails when the close frame
  715. # is empty and parse_close() synthetizes a 1005 close code.
  716. await self.write_close_frame(frame.data)
  717. except ConnectionClosed:
  718. # It doesn't really matter if the connection was closed
  719. # before we could send back a close frame.
  720. pass
  721. return None
  722. elif frame.opcode == OP_PING:
  723. # Answer pings.
  724. ping_hex = frame.data.hex() or "[empty]"
  725. logger.debug(
  726. "%s - received ping, sending pong: %s", self.side, ping_hex
  727. )
  728. await self.pong(frame.data)
  729. elif frame.opcode == OP_PONG:
  730. # Acknowledge pings on solicited pongs.
  731. if frame.data in self.pings:
  732. logger.debug(
  733. "%s - received solicited pong: %s",
  734. self.side,
  735. frame.data.hex() or "[empty]",
  736. )
  737. # Acknowledge all pings up to the one matching this pong.
  738. ping_id = None
  739. ping_ids = []
  740. for ping_id, ping in self.pings.items():
  741. ping_ids.append(ping_id)
  742. if not ping.done():
  743. ping.set_result(None)
  744. if ping_id == frame.data:
  745. break
  746. else: # pragma: no cover
  747. assert False, "ping_id is in self.pings"
  748. # Remove acknowledged pings from self.pings.
  749. for ping_id in ping_ids:
  750. del self.pings[ping_id]
  751. ping_ids = ping_ids[:-1]
  752. if ping_ids:
  753. pings_hex = ", ".join(
  754. ping_id.hex() or "[empty]" for ping_id in ping_ids
  755. )
  756. plural = "s" if len(ping_ids) > 1 else ""
  757. logger.debug(
  758. "%s - acknowledged previous ping%s: %s",
  759. self.side,
  760. plural,
  761. pings_hex,
  762. )
  763. else:
  764. logger.debug(
  765. "%s - received unsolicited pong: %s",
  766. self.side,
  767. frame.data.hex() or "[empty]",
  768. )
  769. # 5.6. Data Frames
  770. else:
  771. return frame
  772. async def read_frame(self, max_size: Optional[int]) -> Frame:
  773. """
  774. Read a single frame from the connection.
  775. """
  776. frame = await Frame.read(
  777. self.reader.readexactly,
  778. mask=not self.is_client,
  779. max_size=max_size,
  780. extensions=self.extensions,
  781. )
  782. logger.debug("%s < %r", self.side, frame)
  783. return frame
  784. async def write_frame(
  785. self, fin: bool, opcode: int, data: bytes, *, _expected_state: int = State.OPEN
  786. ) -> None:
  787. # Defensive assertion for protocol compliance.
  788. if self.state is not _expected_state: # pragma: no cover
  789. raise InvalidState(
  790. f"Cannot write to a WebSocket in the {self.state.name} state"
  791. )
  792. frame = Frame(fin, Opcode(opcode), data)
  793. logger.debug("%s > %r", self.side, frame)
  794. frame.write(
  795. self.transport.write, mask=self.is_client, extensions=self.extensions
  796. )
  797. try:
  798. # drain() cannot be called concurrently by multiple coroutines:
  799. # http://bugs.python.org/issue29930. Remove this lock when no
  800. # version of Python where this bugs exists is supported anymore.
  801. async with self._drain_lock:
  802. # Handle flow control automatically.
  803. await self._drain()
  804. except ConnectionError:
  805. # Terminate the connection if the socket died.
  806. self.fail_connection()
  807. # Wait until the connection is closed to raise ConnectionClosed
  808. # with the correct code and reason.
  809. await self.ensure_open()
  810. async def write_close_frame(self, data: bytes = b"") -> None:
  811. """
  812. Write a close frame if and only if the connection state is OPEN.
  813. This dedicated coroutine must be used for writing close frames to
  814. ensure that at most one close frame is sent on a given connection.
  815. """
  816. # Test and set the connection state before sending the close frame to
  817. # avoid sending two frames in case of concurrent calls.
  818. if self.state is State.OPEN:
  819. # 7.1.3. The WebSocket Closing Handshake is Started
  820. self.state = State.CLOSING
  821. logger.debug("%s - state = CLOSING", self.side)
  822. # 7.1.2. Start the WebSocket Closing Handshake
  823. await self.write_frame(True, OP_CLOSE, data, _expected_state=State.CLOSING)
  824. async def keepalive_ping(self) -> None:
  825. """
  826. Send a Ping frame and wait for a Pong frame at regular intervals.
  827. This coroutine exits when the connection terminates and one of the
  828. following happens:
  829. - :meth:`ping` raises :exc:`ConnectionClosed`, or
  830. - :meth:`close_connection` cancels :attr:`keepalive_ping_task`.
  831. """
  832. if self.ping_interval is None:
  833. return
  834. try:
  835. while True:
  836. await asyncio.sleep(
  837. self.ping_interval,
  838. loop=self.loop if sys.version_info[:2] < (3, 8) else None,
  839. )
  840. # ping() raises CancelledError if the connection is closed,
  841. # when close_connection() cancels self.keepalive_ping_task.
  842. # ping() raises ConnectionClosed if the connection is lost,
  843. # when connection_lost() calls abort_pings().
  844. pong_waiter = await self.ping()
  845. if self.ping_timeout is not None:
  846. try:
  847. await asyncio.wait_for(
  848. pong_waiter,
  849. self.ping_timeout,
  850. loop=self.loop if sys.version_info[:2] < (3, 8) else None,
  851. )
  852. except asyncio.TimeoutError:
  853. logger.debug("%s ! timed out waiting for pong", self.side)
  854. self.fail_connection(1011)
  855. break
  856. # Remove this branch when dropping support for Python < 3.8
  857. # because CancelledError no longer inherits Exception.
  858. except asyncio.CancelledError:
  859. raise
  860. except ConnectionClosed:
  861. pass
  862. except Exception:
  863. logger.warning("Unexpected exception in keepalive ping task", exc_info=True)
  864. async def close_connection(self) -> None:
  865. """
  866. 7.1.1. Close the WebSocket Connection
  867. When the opening handshake succeeds, :meth:`connection_open` starts
  868. this coroutine in a task. It waits for the data transfer phase to
  869. complete then it closes the TCP connection cleanly.
  870. When the opening handshake fails, :meth:`fail_connection` does the
  871. same. There's no data transfer phase in that case.
  872. """
  873. try:
  874. # Wait for the data transfer phase to complete.
  875. if hasattr(self, "transfer_data_task"):
  876. try:
  877. await self.transfer_data_task
  878. except asyncio.CancelledError:
  879. pass
  880. # Cancel the keepalive ping task.
  881. if hasattr(self, "keepalive_ping_task"):
  882. self.keepalive_ping_task.cancel()
  883. # A client should wait for a TCP close from the server.
  884. if self.is_client and hasattr(self, "transfer_data_task"):
  885. if await self.wait_for_connection_lost():
  886. # Coverage marks this line as a partially executed branch.
  887. # I supect a bug in coverage. Ignore it for now.
  888. return # pragma: no cover
  889. logger.debug("%s ! timed out waiting for TCP close", self.side)
  890. # Half-close the TCP connection if possible (when there's no TLS).
  891. if self.transport.can_write_eof():
  892. logger.debug("%s x half-closing TCP connection", self.side)
  893. self.transport.write_eof()
  894. if await self.wait_for_connection_lost():
  895. # Coverage marks this line as a partially executed branch.
  896. # I supect a bug in coverage. Ignore it for now.
  897. return # pragma: no cover
  898. logger.debug("%s ! timed out waiting for TCP close", self.side)
  899. finally:
  900. # The try/finally ensures that the transport never remains open,
  901. # even if this coroutine is canceled (for example).
  902. # If connection_lost() was called, the TCP connection is closed.
  903. # However, if TLS is enabled, the transport still needs closing.
  904. # Else asyncio complains: ResourceWarning: unclosed transport.
  905. if self.connection_lost_waiter.done() and self.transport.is_closing():
  906. return
  907. # Close the TCP connection. Buffers are flushed asynchronously.
  908. logger.debug("%s x closing TCP connection", self.side)
  909. self.transport.close()
  910. if await self.wait_for_connection_lost():
  911. return
  912. logger.debug("%s ! timed out waiting for TCP close", self.side)
  913. # Abort the TCP connection. Buffers are discarded.
  914. logger.debug("%s x aborting TCP connection", self.side)
  915. self.transport.abort()
  916. # connection_lost() is called quickly after aborting.
  917. # Coverage marks this line as a partially executed branch.
  918. # I supect a bug in coverage. Ignore it for now.
  919. await self.wait_for_connection_lost() # pragma: no cover
  920. async def wait_for_connection_lost(self) -> bool:
  921. """
  922. Wait until the TCP connection is closed or ``self.close_timeout`` elapses.
  923. Return ``True`` if the connection is closed and ``False`` otherwise.
  924. """
  925. if not self.connection_lost_waiter.done():
  926. try:
  927. await asyncio.wait_for(
  928. asyncio.shield(self.connection_lost_waiter),
  929. self.close_timeout,
  930. loop=self.loop if sys.version_info[:2] < (3, 8) else None,
  931. )
  932. except asyncio.TimeoutError:
  933. pass
  934. # Re-check self.connection_lost_waiter.done() synchronously because
  935. # connection_lost() could run between the moment the timeout occurs
  936. # and the moment this coroutine resumes running.
  937. return self.connection_lost_waiter.done()
  938. def fail_connection(self, code: int = 1006, reason: str = "") -> None:
  939. """
  940. 7.1.7. Fail the WebSocket Connection
  941. This requires:
  942. 1. Stopping all processing of incoming data, which means cancelling
  943. :attr:`transfer_data_task`. The close code will be 1006 unless a
  944. close frame was received earlier.
  945. 2. Sending a close frame with an appropriate code if the opening
  946. handshake succeeded and the other side is likely to process it.
  947. 3. Closing the connection. :meth:`close_connection` takes care of
  948. this once :attr:`transfer_data_task` exits after being canceled.
  949. (The specification describes these steps in the opposite order.)
  950. """
  951. logger.debug(
  952. "%s ! failing %s WebSocket connection with code %d",
  953. self.side,
  954. self.state.name,
  955. code,
  956. )
  957. # Cancel transfer_data_task if the opening handshake succeeded.
  958. # cancel() is idempotent and ignored if the task is done already.
  959. if hasattr(self, "transfer_data_task"):
  960. self.transfer_data_task.cancel()
  961. # Send a close frame when the state is OPEN (a close frame was already
  962. # sent if it's CLOSING), except when failing the connection because of
  963. # an error reading from or writing to the network.
  964. # Don't send a close frame if the connection is broken.
  965. if code != 1006 and self.state is State.OPEN:
  966. frame_data = serialize_close(code, reason)
  967. # Write the close frame without draining the write buffer.
  968. # Keeping fail_connection() synchronous guarantees it can't
  969. # get stuck and simplifies the implementation of the callers.
  970. # Not drainig the write buffer is acceptable in this context.
  971. # This duplicates a few lines of code from write_close_frame()
  972. # and write_frame().
  973. self.state = State.CLOSING
  974. logger.debug("%s - state = CLOSING", self.side)
  975. frame = Frame(True, OP_CLOSE, frame_data)
  976. logger.debug("%s > %r", self.side, frame)
  977. frame.write(
  978. self.transport.write, mask=self.is_client, extensions=self.extensions
  979. )
  980. # Start close_connection_task if the opening handshake didn't succeed.
  981. if not hasattr(self, "close_connection_task"):
  982. self.close_connection_task = self.loop.create_task(self.close_connection())
  983. def abort_pings(self) -> None:
  984. """
  985. Raise ConnectionClosed in pending keepalive pings.
  986. They'll never receive a pong once the connection is closed.
  987. """
  988. assert self.state is State.CLOSED
  989. exc = self.connection_closed_exc()
  990. for ping in self.pings.values():
  991. ping.set_exception(exc)
  992. # If the exception is never retrieved, it will be logged when ping
  993. # is garbage-collected. This is confusing for users.
  994. # Given that ping is done (with an exception), canceling it does
  995. # nothing, but it prevents logging the exception.
  996. ping.cancel()
  997. if self.pings:
  998. pings_hex = ", ".join(ping_id.hex() or "[empty]" for ping_id in self.pings)
  999. plural = "s" if len(self.pings) > 1 else ""
  1000. logger.debug(
  1001. "%s - aborted pending ping%s: %s", self.side, plural, pings_hex
  1002. )
  1003. # asyncio.Protocol methods
  1004. def connection_made(self, transport: asyncio.BaseTransport) -> None:
  1005. """
  1006. Configure write buffer limits.
  1007. The high-water limit is defined by ``self.write_limit``.
  1008. The low-water limit currently defaults to ``self.write_limit // 4`` in
  1009. :meth:`~asyncio.WriteTransport.set_write_buffer_limits`, which should
  1010. be all right for reasonable use cases of this library.
  1011. This is the earliest point where we can get hold of the transport,
  1012. which means it's the best point for configuring it.
  1013. """
  1014. logger.debug("%s - event = connection_made(%s)", self.side, transport)
  1015. transport = cast(asyncio.Transport, transport)
  1016. transport.set_write_buffer_limits(self.write_limit)
  1017. self.transport = transport
  1018. # Copied from asyncio.StreamReaderProtocol
  1019. self.reader.set_transport(transport)
  1020. def connection_lost(self, exc: Optional[Exception]) -> None:
  1021. """
  1022. 7.1.4. The WebSocket Connection is Closed.
  1023. """
  1024. logger.debug("%s - event = connection_lost(%s)", self.side, exc)
  1025. self.state = State.CLOSED
  1026. logger.debug("%s - state = CLOSED", self.side)
  1027. if not hasattr(self, "close_code"):
  1028. self.close_code = 1006
  1029. if not hasattr(self, "close_reason"):
  1030. self.close_reason = ""
  1031. logger.debug(
  1032. "%s x code = %d, reason = %s",
  1033. self.side,
  1034. self.close_code,
  1035. self.close_reason or "[no reason]",
  1036. )
  1037. self.abort_pings()
  1038. # If self.connection_lost_waiter isn't pending, that's a bug, because:
  1039. # - it's set only here in connection_lost() which is called only once;
  1040. # - it must never be canceled.
  1041. self.connection_lost_waiter.set_result(None)
  1042. if True: # pragma: no cover
  1043. # Copied from asyncio.StreamReaderProtocol
  1044. if self.reader is not None:
  1045. if exc is None:
  1046. self.reader.feed_eof()
  1047. else:
  1048. self.reader.set_exception(exc)
  1049. # Copied from asyncio.FlowControlMixin
  1050. # Wake up the writer if currently paused.
  1051. if not self._paused:
  1052. return
  1053. waiter = self._drain_waiter
  1054. if waiter is None:
  1055. return
  1056. self._drain_waiter = None
  1057. if waiter.done():
  1058. return
  1059. if exc is None:
  1060. waiter.set_result(None)
  1061. else:
  1062. waiter.set_exception(exc)
  1063. def pause_writing(self) -> None: # pragma: no cover
  1064. assert not self._paused
  1065. self._paused = True
  1066. def resume_writing(self) -> None: # pragma: no cover
  1067. assert self._paused
  1068. self._paused = False
  1069. waiter = self._drain_waiter
  1070. if waiter is not None:
  1071. self._drain_waiter = None
  1072. if not waiter.done():
  1073. waiter.set_result(None)
  1074. def data_received(self, data: bytes) -> None:
  1075. logger.debug("%s - event = data_received(<%d bytes>)", self.side, len(data))
  1076. self.reader.feed_data(data)
  1077. def eof_received(self) -> None:
  1078. """
  1079. Close the transport after receiving EOF.
  1080. The WebSocket protocol has its own closing handshake: endpoints close
  1081. the TCP or TLS connection after sending and receiving a close frame.
  1082. As a consequence, they never need to write after receiving EOF, so
  1083. there's no reason to keep the transport open by returning ``True``.
  1084. Besides, that doesn't work on TLS connections.
  1085. """
  1086. logger.debug("%s - event = eof_received()", self.side)
  1087. self.reader.feed_eof()