Add trio connection.

Author: aaugustin

docs/topics/keepalive.rst

@@ -136,8 +136,8 @@ measured during the last exchange of Ping and Pong frames::
 Alternatively, you can measure the latency at any time by calling
 :attr:`~asyncio.connection.Connection.ping` and awaiting its result::
 
-    pong_waiter = await websocket.ping()
-    latency = await pong_waiter
+    pong_received = await websocket.ping()
+    latency = await pong_received
 
 Latency between a client and a server may increase for two reasons:
 

src/websockets/asyncio/connection.py

@@ -101,10 +101,10 @@ def __init__(
         self.close_deadline: float | None = None
 
         # Protect sending fragmented messages.
-        self.fragmented_send_waiter: asyncio.Future[None] | None = None
+        self.send_in_progress: asyncio.Future[None] | None = None
 
         # Mapping of ping IDs to pong waiters, in chronological order.
-        self.pong_waiters: dict[bytes, tuple[asyncio.Future[float], float]] = {}
+        self.pending_pings: dict[bytes, tuple[asyncio.Future[float], float]] = {}
 
         self.latency: float = 0
         """
@@ -468,8 +468,8 @@ async def send(
         """
         # While sending a fragmented message, prevent sending other messages
         # until all fragments are sent.
-        while self.fragmented_send_waiter is not None:
-            await asyncio.shield(self.fragmented_send_waiter)
+        while self.send_in_progress is not None:
+            await asyncio.shield(self.send_in_progress)
 
         # Unfragmented message -- this case must be handled first because
         # strings and bytes-like objects are iterable.
@@ -502,8 +502,8 @@ async def send(
             except StopIteration:
                 return
 
-            assert self.fragmented_send_waiter is None
-            self.fragmented_send_waiter = self.loop.create_future()
+            assert self.send_in_progress is None
+            self.send_in_progress = self.loop.create_future()
             try:
                 # First fragment.
                 if isinstance(chunk, str):
@@ -549,8 +549,8 @@ async def send(
                 raise
 
             finally:
-                self.fragmented_send_waiter.set_result(None)
-                self.fragmented_send_waiter = None
+                self.send_in_progress.set_result(None)
+                self.send_in_progress = None
 
         # Fragmented message -- async iterator.
 
@@ -561,8 +561,8 @@ async def send(
             except StopAsyncIteration:
                 return
 
-            assert self.fragmented_send_waiter is None
-            self.fragmented_send_waiter = self.loop.create_future()
+            assert self.send_in_progress is None
+            self.send_in_progress = self.loop.create_future()
             try:
                 # First fragment.
                 if isinstance(chunk, str):
@@ -610,8 +610,8 @@ async def send(
                 raise
 
             finally:
-                self.fragmented_send_waiter.set_result(None)
-                self.fragmented_send_waiter = None
+                self.send_in_progress.set_result(None)
+                self.send_in_progress = None
 
         else:
             raise TypeError("data must be str, bytes, iterable, or async iterable")
@@ -639,7 +639,7 @@ async def close(
             # The context manager takes care of waiting for the TCP connection
             # to terminate after calling a method that sends a close frame.
             async with self.send_context():
-                if self.fragmented_send_waiter is not None:
+                if self.send_in_progress is not None:
                     self.protocol.fail(
                         CloseCode.INTERNAL_ERROR,
                         "close during fragmented message",
@@ -681,9 +681,9 @@ async def ping(self, data: DataLike | None = None) -> Awaitable[float]:
 
             ::
 
-                pong_waiter = await ws.ping()
+                pong_received = await ws.ping()
                 # only if you want to wait for the corresponding pong
-                latency = await pong_waiter
+                latency = await pong_received
 
         Raises:
             ConnectionClosed: When the connection is closed.
@@ -700,19 +700,19 @@ async def ping(self, data: DataLike | None = None) -> Awaitable[float]:
 
         async with self.send_context():
             # Protect against duplicates if a payload is explicitly set.
-            if data in self.pong_waiters:
+            if data in self.pending_pings:
                 raise ConcurrencyError("already waiting for a pong with the same data")
 
             # Generate a unique random payload otherwise.
-            while data is None or data in self.pong_waiters:
+            while data is None or data in self.pending_pings:
                 data = struct.pack("!I", random.getrandbits(32))
 
-            pong_waiter = self.loop.create_future()
+            pong_received = self.loop.create_future()
             # The event loop's default clock is time.monotonic(). Its resolution
             # is a bit low on Windows (~16ms). This is improved in Python 3.13.
-            self.pong_waiters[data] = (pong_waiter, self.loop.time())
+            self.pending_pings[data] = (pong_received, self.loop.time())
             self.protocol.send_ping(data)
-            return pong_waiter
+            return pong_received
 
     async def pong(self, data: DataLike = b"") -> None:
         """
@@ -761,7 +761,7 @@ def acknowledge_pings(self, data: bytes) -> None:
 
         """
         # Ignore unsolicited pong.
-        if data not in self.pong_waiters:
+        if data not in self.pending_pings:
             return
 
         pong_timestamp = self.loop.time()
@@ -770,20 +770,20 @@ def acknowledge_pings(self, data: bytes) -> None:
         # Acknowledge all previous pings too in that case.
         ping_id = None
         ping_ids = []
-        for ping_id, (pong_waiter, ping_timestamp) in self.pong_waiters.items():
+        for ping_id, (pong_received, ping_timestamp) in self.pending_pings.items():
             ping_ids.append(ping_id)
             latency = pong_timestamp - ping_timestamp
-            if not pong_waiter.done():
-                pong_waiter.set_result(latency)
+            if not pong_received.done():
+                pong_received.set_result(latency)
             if ping_id == data:
                 self.latency = latency
                 break
         else:
             raise AssertionError("solicited pong not found in pings")
 
-        # Remove acknowledged pings from self.pong_waiters.
+        # Remove acknowledged pings from self.pending_pings.
         for ping_id in ping_ids:
-            del self.pong_waiters[ping_id]
+            del self.pending_pings[ping_id]
 
     def abort_pings(self) -> None:
         """
@@ -795,16 +795,16 @@ def abort_pings(self) -> None:
         assert self.protocol.state is CLOSED
         exc = self.protocol.close_exc
 
-        for pong_waiter, _ping_timestamp in self.pong_waiters.values():
-            if not pong_waiter.done():
-                pong_waiter.set_exception(exc)
+        for pong_received, _ping_timestamp in self.pending_pings.values():
+            if not pong_received.done():
+                pong_received.set_exception(exc)
             # If the exception is never retrieved, it will be logged when ping
             # is garbage-collected. This is confusing for users.
             # Given that ping is done (with an exception), canceling it does
             # nothing, but it prevents logging the exception.
-            pong_waiter.cancel()
+            pong_received.cancel()
 
-        self.pong_waiters.clear()
+        self.pending_pings.clear()
 
     async def keepalive(self) -> None:
         """
@@ -825,7 +825,7 @@ async def keepalive(self) -> None:
                 # connection to be closed before raising ConnectionClosed.
                 # However, connection_lost() cancels keepalive_task before
                 # it gets a chance to resume excuting.
-                pong_waiter = await self.ping()
+                pong_received = await self.ping()
                 if self.debug:
                     self.logger.debug("% sent keepalive ping")
 
@@ -834,9 +834,9 @@ async def keepalive(self) -> None:
                         async with asyncio_timeout(self.ping_timeout):
                             # connection_lost cancels keepalive immediately
                             # after setting a ConnectionClosed exception on
-                            # pong_waiter. A CancelledError is raised here,
+                            # pong_received. A CancelledError is raised here,
                             # not a ConnectionClosed exception.
-                            latency = await pong_waiter
+                            latency = await pong_received
                         self.logger.debug("% received keepalive pong")
                     except asyncio.TimeoutError:
                         if self.debug:
@@ -1205,7 +1205,7 @@ def broadcast(
         if connection.protocol.state is not OPEN:
             continue
 
-        if connection.fragmented_send_waiter is not None:
+        if connection.send_in_progress is not None:
             if raise_exceptions:
                 exception = ConcurrencyError("sending a fragmented message")
                 exceptions.append(exception)

src/websockets/protocol.py

@@ -254,7 +254,7 @@ def close_exc(self) -> ConnectionClosed:
 
     # Public methods for receiving data.
 
-    def receive_data(self, data: bytes) -> None:
+    def receive_data(self, data: bytes | bytearray) -> None:
         """
         Receive data from the network.
 

src/websockets/streams.py

@@ -112,7 +112,7 @@ def at_eof(self) -> Generator[None, None, bool]:
             # tell if until either feed_data() or feed_eof() is called.
             yield
 
-    def feed_data(self, data: bytes) -> None:
+    def feed_data(self, data: bytes | bytearray) -> None:
         """
         Write data to the stream.
 

src/websockets/sync/connection.py

@@ -104,7 +104,7 @@ def __init__(
         self.send_in_progress = False
 
         # Mapping of ping IDs to pong waiters, in chronological order.
-        self.pong_waiters: dict[bytes, tuple[threading.Event, float, bool]] = {}
+        self.pending_pings: dict[bytes, tuple[threading.Event, float, bool]] = {}
 
         self.latency: float = 0
         """
@@ -633,8 +633,9 @@ def ping(
 
             ::
 
-                pong_event = ws.ping()
-                pong_event.wait()  # only if you want to wait for the pong
+                pong_received = ws.ping()
+                # only if you want to wait for the corresponding pong
+                pong_received.wait()
 
         Raises:
             ConnectionClosed: When the connection is closed.
@@ -651,17 +652,17 @@ def ping(
 
         with self.send_context():
             # Protect against duplicates if a payload is explicitly set.
-            if data in self.pong_waiters:
+            if data in self.pending_pings:
                 raise ConcurrencyError("already waiting for a pong with the same data")
 
             # Generate a unique random payload otherwise.
-            while data is None or data in self.pong_waiters:
+            while data is None or data in self.pending_pings:
                 data = struct.pack("!I", random.getrandbits(32))
 
-            pong_waiter = threading.Event()
-            self.pong_waiters[data] = (pong_waiter, time.monotonic(), ack_on_close)
+            pong_received = threading.Event()
+            self.pending_pings[data] = (pong_received, time.monotonic(), ack_on_close)
             self.protocol.send_ping(data)
-            return pong_waiter
+            return pong_received
 
     def pong(self, data: DataLike = b"") -> None:
         """
@@ -711,7 +712,7 @@ def acknowledge_pings(self, data: bytes) -> None:
         """
         with self.protocol_mutex:
             # Ignore unsolicited pong.
-            if data not in self.pong_waiters:
+            if data not in self.pending_pings:
                 return
 
             pong_timestamp = time.monotonic()
@@ -721,21 +722,21 @@ def acknowledge_pings(self, data: bytes) -> None:
             ping_id = None
             ping_ids = []
             for ping_id, (
-                pong_waiter,
+                pong_received,
                 ping_timestamp,
                 _ack_on_close,
-            ) in self.pong_waiters.items():
+            ) in self.pending_pings.items():
                 ping_ids.append(ping_id)
-                pong_waiter.set()
+                pong_received.set()
                 if ping_id == data:
                     self.latency = pong_timestamp - ping_timestamp
                     break
             else:
                 raise AssertionError("solicited pong not found in pings")
 
-            # Remove acknowledged pings from self.pong_waiters.
+            # Remove acknowledged pings from self.pending_pings.
             for ping_id in ping_ids:
-                del self.pong_waiters[ping_id]
+                del self.pending_pings[ping_id]
 
     def acknowledge_pending_pings(self) -> None:
         """
@@ -744,11 +745,11 @@ def acknowledge_pending_pings(self) -> None:
         """
         assert self.protocol.state is CLOSED
 
-        for pong_waiter, _ping_timestamp, ack_on_close in self.pong_waiters.values():
+        for pong_received, _ping_timestamp, ack_on_close in self.pending_pings.values():
             if ack_on_close:
-                pong_waiter.set()
+                pong_received.set()
 
-        self.pong_waiters.clear()
+        self.pending_pings.clear()
 
     def keepalive(self) -> None:
         """
@@ -766,15 +767,14 @@ def keepalive(self) -> None:
                     break
 
                 try:
-                    pong_waiter = self.ping(ack_on_close=True)
+                    pong_received = self.ping(ack_on_close=True)
                 except ConnectionClosed:
                     break
                 if self.debug:
                     self.logger.debug("% sent keepalive ping")
 
                 if self.ping_timeout is not None:
-                    #
-                    if pong_waiter.wait(self.ping_timeout):
+                    if pong_received.wait(self.ping_timeout):
                         if self.debug:
                             self.logger.debug("% received keepalive pong")
                     else:
@@ -808,7 +808,7 @@ def recv_events(self) -> None:
 
         Run this method in a thread as long as the connection is alive.
 
-        ``recv_events()`` exits immediately when the ``self.socket`` is closed.
+        ``recv_events()`` exits immediately when ``self.socket`` is closed.
 
         """
         try:
@@ -983,6 +983,7 @@ def send_context(
                 # Minor layering violation: we assume that the connection
                 # will be closing soon if it isn't in the expected state.
                 wait_for_close = True
+                # TODO: calculate close deadline if not set?
                 raise_close_exc = True
 
         # To avoid a deadlock, release the connection lock by exiting the