Maybe resume read nowait

CHANGES.rst

@@ -6,7 +6,8 @@ CHANGES
 
 - Make CookieJar compatible with 32-bit systems #1188
 
--
+- Fix FlowControlStreamReader.read_nowait so that it checks
+  whether the transport is paused
 
 -
 

CONTRIBUTORS.txt

@@ -3,6 +3,7 @@ Contributors
 
 A. Jesse Jiryu Davis
 Alejandro Gómez
+Aleksandr Danshyn
 Aleksey Kutepov
 Alex Khomchenko
 Alex Lisovoy

aiohttp/streams.py

@@ -501,29 +501,19 @@ def read(self):
 
 def maybe_resume(func):
 
-    @asyncio.coroutine
-    @functools.wraps(func)
-    def wrapper(self, *args, **kw):
-        result = yield from func(self, *args, **kw)
-
-        if self._stream.paused:
-            if self._buffer_size < self._b_limit:
-                try:
-                    self._stream.transport.resume_reading()
-                except (AttributeError, NotImplementedError):
-                    pass
-                else:
-                    self._stream.paused = False
-        else:
-            if self._buffer_size > self._b_limit:
-                try:
-                    self._stream.transport.pause_reading()
-                except (AttributeError, NotImplementedError):
-                    pass
-                else:
-                    self._stream.paused = True
-
-        return result
+    if asyncio.iscoroutinefunction(func):
+        @asyncio.coroutine
+        @functools.wraps(func)
+        def wrapper(self, *args, **kw):
+            result = yield from func(self, *args, **kw)
+            self._check_buffer_size()
+            return result
+    else:
+        @functools.wraps(func)
+        def wrapper(self, *args, **kw):
+            result = func(self, *args, **kw)
+            self._check_buffer_size()
+            return result
 
     return wrapper
 
@@ -542,6 +532,26 @@ def __init__(self, stream, limit=DEFAULT_LIMIT, *args, **kwargs):
                 self._stream.transport.resume_reading()
             except (AttributeError, NotImplementedError):
                 pass
+            else:
+                self._stream.paused = False
+
+    def _check_buffer_size(self):
+        if self._stream.paused:
+            if self._buffer_size < self._b_limit:
+                try:
+                    self._stream.transport.resume_reading()
+                except (AttributeError, NotImplementedError):
+                    pass
+                else:
+                    self._stream.paused = False
+        else:
+            if self._buffer_size > self._b_limit:
+                try:
+                    self._stream.transport.pause_reading()
+                except (AttributeError, NotImplementedError):
+                    pass
+                else:
+                    self._stream.paused = True
 
     def feed_data(self, data, size=0):
         has_waiter = self._waiter is not None and not self._waiter.cancelled()
@@ -558,21 +568,29 @@ def feed_data(self, data, size=0):
                 self._stream.paused = True
 
     @maybe_resume
+    @asyncio.coroutine
     def read(self, n=-1):
         return (yield from super().read(n))
 
     @maybe_resume
+    @asyncio.coroutine
     def readline(self):
         return (yield from super().readline())
 
     @maybe_resume
+    @asyncio.coroutine
     def readany(self):
         return (yield from super().readany())
 
     @maybe_resume
+    @asyncio.coroutine
     def readexactly(self, n):
         return (yield from super().readexactly(n))
 
+    @maybe_resume
+    def read_nowait(self, n=-1):
+        return super().read_nowait(n)
+
 
 class FlowControlDataQueue(DataQueue):
     """FlowControlDataQueue resumes and pauses an underlying stream.

tests/test_flowcontrol_streams.py

@@ -8,7 +8,7 @@
 class TestFlowControlStreamReader(unittest.TestCase):
 
     def setUp(self):
-        self.stream = mock.Mock()
+        self.stream = mock.Mock(paused=False)
         self.transp = self.stream.transport
         self.loop = asyncio.new_event_loop()
         asyncio.set_event_loop(None)
@@ -22,34 +22,34 @@ def _make_one(self, *args, **kwargs):
 
     def test_read(self):
         r = self._make_one()
-        r.paused = True
+        r._stream.paused = True
         r.feed_data(b'da', 2)
         res = self.loop.run_until_complete(r.read(1))
         self.assertEqual(res, b'd')
         self.assertTrue(self.transp.resume_reading.called)
 
     def test_readline(self):
         r = self._make_one()
-        r.paused = True
+        r._stream.paused = True
         r.feed_data(b'data\n', 5)
         res = self.loop.run_until_complete(r.readline())
         self.assertEqual(res, b'data\n')
         self.assertTrue(self.transp.resume_reading.called)
 
     def test_readany(self):
         r = self._make_one()
-        r.paused = True
+        r._stream.paused = True
         r.feed_data(b'data', 4)
         res = self.loop.run_until_complete(r.readany())
         self.assertEqual(res, b'data')
         self.assertTrue(self.transp.resume_reading.called)
 
     def test_readexactly(self):
         r = self._make_one()
-        r.paused = True
-        r.feed_data(b'datadata', 8)
-        res = self.loop.run_until_complete(r.readexactly(2))
-        self.assertEqual(res, b'da')
+        r._stream.paused = True
+        r.feed_data(b'data', 4)
+        res = self.loop.run_until_complete(r.readexactly(3))
+        self.assertEqual(res, b'dat')
         self.assertTrue(self.transp.resume_reading.called)
 
     def test_feed_data(self):
@@ -58,6 +58,68 @@ def test_feed_data(self):
         r.feed_data(b'datadata', 8)
         self.assertTrue(self.transp.pause_reading.called)
 
+    def test_read_nowait(self):
+        r = self._make_one()
+        r._stream.paused = False
+        r.feed_data(b'data1', 5)
+        r.feed_data(b'data2', 5)
+        r.feed_data(b'data3', 5)
+        self.assertTrue(self.stream.paused)
+
+        res = self.loop.run_until_complete(r.read(5))
+        self.assertTrue(res == b'data1')
+        # _buffer_size > _buffer_limit
+        self.assertTrue(self.transp.pause_reading.call_count == 1)
+        self.assertTrue(self.transp.resume_reading.call_count == 0)
+        self.assertTrue(self.stream.paused)
+
+        r._stream.paused = False
+        res = r.read_nowait(5)
+        self.assertTrue(res == b'data2')
+        # _buffer_size > _buffer_limit
+        self.assertTrue(self.transp.pause_reading.call_count == 2)
+        self.assertTrue(self.transp.resume_reading.call_count == 0)
+        self.assertTrue(self.stream.paused)
+
+        res = r.read_nowait(5)
+        self.assertTrue(res == b'data3')
+        # _buffer_size < _buffer_limit
+        self.assertTrue(self.transp.pause_reading.call_count == 2)
+        self.assertTrue(self.transp.resume_reading.call_count == 1)
+        self.assertTrue(not self.stream.paused)
+
+        res = r.read_nowait(5)
+        self.assertTrue(res == b'')
+        # _buffer_size < _buffer_limit
+        self.assertTrue(self.transp.pause_reading.call_count == 2)
+        self.assertTrue(self.transp.resume_reading.call_count == 1)
+        self.assertTrue(not self.stream.paused)
+
+    def test_rudimentary_transport(self):
+        self.transp.resume_reading.side_effect = NotImplementedError()
+        self.transp.pause_reading.side_effect = NotImplementedError()
+        self.stream.paused = True
+
+        r = self._make_one()
+        self.assertTrue(self.transp.pause_reading.call_count == 0)
+        self.assertTrue(self.transp.resume_reading.call_count == 1)
+        self.assertTrue(self.stream.paused)
+
+        r.feed_data(b'data', 4)
+        res = self.loop.run_until_complete(r.read(4))
+        self.assertTrue(self.transp.pause_reading.call_count == 0)
+        self.assertTrue(self.transp.resume_reading.call_count == 2)
+        self.assertTrue(self.stream.paused)
+        self.assertTrue(res == b'data')
+
+        self.stream.paused = False
+        r.feed_data(b'data', 4)
+        res = self.loop.run_until_complete(r.read(1))
+        self.assertTrue(self.transp.pause_reading.call_count == 2)
+        self.assertTrue(self.transp.resume_reading.call_count == 2)
+        self.assertTrue(not self.stream.paused)
+        self.assertTrue(res == b'd')
+
 
 class FlowControlMixin: