Stream unpauses protocol before releasing connection (#10171)

CHANGES/10169.bugfix.rst

@@ -0,0 +1,3 @@
+Fixed a hang where a connection previously used for a streaming
+download could be returned to the pool in a paused state.
+-- by :user:`javitonino`.

CONTRIBUTORS.txt

@@ -179,6 +179,7 @@ Jan Buchar
 Jan Gosmann
 Jarno Elonen
 Jashandeep Sohi
+Javier Torres
 Jean-Baptiste Estival
 Jens Steinhauser
 Jeonghun Lee

aiohttp/streams.py

@@ -220,6 +220,9 @@ def feed_eof(self) -> None:
             self._eof_waiter = None
             set_result(waiter, None)
 
+        if self._protocol._reading_paused:
+            self._protocol.resume_reading()
+
         for cb in self._eof_callbacks:
             try:
                 cb()

tests/test_flowcontrol_streams.py

@@ -107,3 +107,25 @@ async def test_read_nowait(self, stream: streams.StreamReader) -> None:
         res = stream.read_nowait(5)
         assert res == b""
         assert stream._protocol.resume_reading.call_count == 1  # type: ignore[attr-defined]
+
+    async def test_resumed_on_eof(self, stream: streams.StreamReader) -> None:
+        stream.feed_data(b"data")
+        assert stream._protocol.pause_reading.call_count == 1  # type: ignore[attr-defined]
+        assert stream._protocol.resume_reading.call_count == 0  # type: ignore[attr-defined]
+        stream._protocol._reading_paused = True
+
+        stream.feed_eof()
+        assert stream._protocol.resume_reading.call_count == 1  # type: ignore[attr-defined]
+
+
+async def test_stream_reader_eof_when_full() -> None:
+    loop = asyncio.get_event_loop()
+    protocol = BaseProtocol(loop=loop)
+    protocol.transport = asyncio.Transport()
+    stream = streams.StreamReader(protocol, 1024, loop=loop)
+
+    data_len = stream._high_water + 1
+    stream.feed_data(b"0" * data_len)
+    assert protocol._reading_paused
+    stream.feed_eof()
+    assert not protocol._reading_paused