Add wait_closed to the threading implementation.

docs/project/changelog.rst

@@ -32,6 +32,12 @@ notice.
 
 *In development*
 
+New features
+............
+
+* Added :meth:`~sync.connection.Connection.wait_closed` to the :mod:`threading`
+  implementation.
+
 .. _17.1:
 
 17.1

docs/reference/sync/client.rst

@@ -29,6 +29,8 @@ Using a connection
 
     .. automethod:: close
 
+    .. automethod:: wait_closed
+
     .. automethod:: ping
 
     .. automethod:: pong

docs/reference/sync/connection.rst

@@ -17,6 +17,8 @@ Both sides (:mod:`threading`)
 
     .. automethod:: close
 
+    .. automethod:: wait_closed
+
     .. automethod:: ping
 
     .. automethod:: pong

docs/reference/sync/server.rst

@@ -53,6 +53,8 @@ Using a connection
 
     .. automethod:: close
 
+    .. automethod:: wait_closed
+
     .. automethod:: ping
 
     .. automethod:: pong

src/websockets/sync/connection.py

@@ -612,6 +612,26 @@ def close(
             # They mean that the connection is closed, which was the goal.
             pass
 
+    def wait_closed(self, timeout: float | None = None) -> bool:
+        """
+        Wait until the connection is closed.
+
+        :meth:`wait_closed` waits for the closing handshake to complete and for
+        the TCP connection to terminate.
+
+        If a timeout is provided and elapses before the connection is closed,
+        :meth:`wait_closed` returns :obj:`False`. Else, it returns :obj:`True`.
+
+        Args:
+            timeout: Optional timeout in seconds.
+
+        Returns:
+            Whether the connection is closed.
+
+        """
+        self.recv_events_thread.join(timeout)
+        return not self.recv_events_thread.is_alive()
+
     def ping(
         self,
         data: DataLike | None = None,

tests/asyncio/test_connection.py

@@ -929,11 +929,11 @@ async def fragments():
 
     async def test_wait_closed(self):
         """wait_closed waits for the connection to close."""
-        wait_closed_task = asyncio.create_task(self.connection.wait_closed())
-        await asyncio.sleep(0)  # let the event loop start wait_closed_task
-        self.assertFalse(wait_closed_task.done())
+        waiter_task = asyncio.create_task(self.connection.wait_closed())
+        await asyncio.sleep(0)  # let the event loop start waiter_task
+        self.assertFalse(waiter_task.done())
         await self.connection.close()
-        self.assertTrue(wait_closed_task.done())
+        self.assertTrue(waiter_task.done())
 
     # Test ping.
 

tests/sync/test_connection.py

@@ -665,6 +665,35 @@ def fragments():
         )
         self.assertIsNone(exc.__cause__)
 
+    # Test wait_closed.
+
+    def test_wait_closed(self):
+        """wait_closed waits for the connection to close."""
+        closed = threading.Event()
+
+        def waiter():
+            self.connection.wait_closed()
+            closed.set()
+
+        with self.run_in_thread(waiter):
+            self.assertFalse(closed.wait(MS))
+            self.connection.close()
+            self.assertTrue(closed.wait(MS))
+
+    def test_wait_closed_no_timeout(self):
+        """wait_closed without a timeout returns True."""
+        self.connection.close()
+        self.assertTrue(self.connection.wait_closed())
+
+    def test_wait_closed_with_timeout(self):
+        """wait_closed returns True when the connection closes."""
+        self.connection.close()
+        self.assertTrue(self.connection.wait_closed(timeout=MS))
+
+    def test_wait_closed_with_timeout_elapsed(self):
+        """wait_closed returns False when the timeout elapses."""
+        self.assertFalse(self.connection.wait_closed(timeout=MS))
+
     # Test ping.
 
     @patch("random.getrandbits")

tests/trio/test_connection.py

@@ -863,11 +863,11 @@ async def test_wait_closed(self):
         """wait_closed waits for the connection to close."""
         closed = trio.Event()
 
-        async def closer():
+        async def waiter():
             await self.connection.wait_closed()
             closed.set()
 
-        self.nursery.start_soon(closer)
+        self.nursery.start_soon(waiter)
         await trio.testing.wait_all_tasks_blocked()
         self.assertFalse(closed.is_set())