Fixing pause_reading called in connection made is ignored

added is_reading() for unix transports

Lib/asyncio/selector_events.py

@@ -728,9 +728,8 @@ class _SelectorTransport(transports._FlowControlMixin,
         return len(self._buffer)
 
     def _add_reader(self, fd, callback, *args):
-        if self._closing:
+        if not self.is_reading():
             return
-
         self._loop._add_reader(fd, callback, *args)
 
 

Lib/asyncio/unix_events.py

@@ -460,13 +460,18 @@ class _UnixReadPipeTransport(transports.ReadTransport):
 
         self._loop.call_soon(self._protocol.connection_made, self)
         # only start reading when connection_made() has been called
-        self._loop.call_soon(self._loop._add_reader,
+        self._loop.call_soon(self._add_reader,
                              self._fileno, self._read_ready)
         if waiter is not None:
             # only wake up the waiter when connection_made() has been called
             self._loop.call_soon(futures._set_result_unless_cancelled,
                                  waiter, None)
 
+    def _add_reader(self, fd, callback, *args):
+        if not self.is_reading():
+            return
+        self._loop._add_reader(fd, callback, *args)
+
     def __repr__(self):
         info = [self.__class__.__name__]
         if self._pipe is None:
@@ -531,6 +536,9 @@ class _UnixReadPipeTransport(transports.ReadTransport):
     def is_closing(self):
         return self._closing
 
+    def is_reading(self):
+        return not self._paused and not self._closing
+
     def close(self):
         if not self._closing:
             self._close(None)
@@ -601,7 +609,7 @@ class _UnixWritePipeTransport(transports._FlowControlMixin,
         # works for pipes and sockets. (Exception: OS X 10.4?  Issue #19294.)
         if is_socket or (is_fifo and not sys.platform.startswith("aix")):
             # only start reading when connection_made() has been called
-            self._loop.call_soon(self._loop._add_reader,
+            self._loop.call_soon(self._add_reader,
                                  self._fileno, self._read_ready)
 
         if waiter is not None:
@@ -609,6 +617,11 @@ class _UnixWritePipeTransport(transports._FlowControlMixin,
             self._loop.call_soon(futures._set_result_unless_cancelled,
                                  waiter, None)
 
+    def _add_reader(self, fd, callback, *args):
+        if not self.is_reading():
+            return
+        self._loop._add_reader(fd, callback, *args)
+
     def __repr__(self):
         info = [self.__class__.__name__]
         if self._pipe is None:
@@ -729,6 +742,9 @@ class _UnixWritePipeTransport(transports._FlowControlMixin,
     def is_closing(self):
         return self._closing
 
+    def is_reading(self):
+        return not self._paused and not self._closing
+
     def close(self):
         if self._pipe is not None and not self._closing:
             # write_eof is all what we needed to close the write pipe

Lib/test/test_asyncio/test_selector_events.py

@@ -554,6 +554,18 @@ class SelectorSocketTransportTests(test_utils.TestCase):
         self.assertFalse(tr.is_reading())
         self.loop.assert_no_reader(7)
 
+    def test_pause_reading_connection_made(self):
+        tr = self.socket_transport()
+        tr.pause_reading()
+        test_utils.run_briefly(self.loop)
+        self.assertFalse(tr.is_reading())
+        self.loop.assert_no_reader(7, tr._read_ready)
+
+        tr.resume_reading()
+        self.assertTrue(tr.is_reading())
+        self.loop.assert_reader(7, tr._read_ready)
+
+
     def test_read_eof_received_error(self):
         transport = self.socket_transport()
         transport.close = mock.Mock()