Prevent close() from blocking when reading is paused.

Fix #1555.

Author: aaugustin

docs/project/changelog.rst

@@ -35,15 +35,18 @@ notice.
 Bug fixes
 .........
 
-* Fixed ``connection.recv(timeout=0)`` in the :mod:`threading` implementation.
-  If a message is already received, it is returned. Previously,
-  :exc:`TimeoutError` was raised incorrectly.
-
 * Wrapped errors when reading the opening handshake request or response in
   :exc:`~exceptions.InvalidMessage` so that :func:`~asyncio.client.connect`
   raises :exc:`~exceptions.InvalidHandshake` or a subclass when the opening
   handshake fails.
 
+* Fixed :meth:`~sync.connection.Connection.recv` with ``timeout=0`` in the
+  :mod:`threading` implementation. If a message is already received, it is
+  returned. Previously, :exc:`TimeoutError` was raised incorrectly.
+
+* Prevented :meth:`~sync.connection.Connection.close` from blocking when
+  receive buffers are saturated in the :mod:`threading` implementation.
+
 .. _14.1:
 
 14.1

src/websockets/asyncio/messages.py

@@ -283,7 +283,7 @@ def close(self) -> None:
         """
         End the stream of frames.
 
-        Callling :meth:`close` concurrently with :meth:`get`, :meth:`get_iter`,
+        Calling :meth:`close` concurrently with :meth:`get`, :meth:`get_iter`,
         or :meth:`put` is safe. They will raise :exc:`EOFError`.
 
         """

src/websockets/sync/messages.py

@@ -298,7 +298,7 @@ def close(self) -> None:
         """
         End the stream of frames.
 
-        Callling :meth:`close` concurrently with :meth:`get`, :meth:`get_iter`,
+        Calling :meth:`close` concurrently with :meth:`get`, :meth:`get_iter`,
         or :meth:`put` is safe. They will raise :exc:`EOFError`.
 
         """
@@ -311,3 +311,8 @@ def close(self) -> None:
             if self.get_in_progress:
                 # Unblock get() or get_iter().
                 self.frames.put(None)
+
+            if self.paused:
+                # Unblock recv_events().
+                self.paused = False
+                self.resume()

tests/sync/test_messages.py

@@ -496,6 +496,18 @@ def test_put_fails_after_close(self):
         with self.assertRaises(EOFError):
             self.assembler.put(Frame(OP_TEXT, b"caf\xc3\xa9"))
 
+    def test_close_resumes_reading(self):
+        """close unblocks reading when queue is above the high-water mark."""
+        self.assembler.put(Frame(OP_TEXT, b"caf\xc3\xa9"))
+        self.assembler.put(Frame(OP_TEXT, b"more caf\xc3\xa9"))
+        self.assembler.put(Frame(OP_TEXT, b"water"))
+
+        # queue is at the high-water mark
+        assert self.assembler.paused
+
+        self.assembler.close()
+        self.resume.assert_called_once_with()
+
     def test_close_is_idempotent(self):
         """close can be called multiple times safely."""
         self.assembler.close()