self._fileno = pipe.fileno()
self._protocol = protocol
self._closing = False
+ self._paused = False
mode = os.fstat(self._fileno).st_mode
if not (stat.S_ISFIFO(mode) or
self._loop.call_soon(self._call_connection_lost, None)
def pause_reading(self):
+ if self._closing or self._paused:
+ return
+ self._paused = True
self._loop._remove_reader(self._fileno)
+ if self._loop.get_debug():
+ logger.debug("%r pauses reading", self)
def resume_reading(self):
+ if self._closing or not self._paused:
+ return
+ self._paused = False
self._loop._add_reader(self._fileno, self._read_ready)
+ if self._loop.get_debug():
+ logger.debug("%r resumes reading", self)
def set_protocol(self, protocol):
self._protocol = protocol
@mock.patch('os.read')
def test_resume_reading(self, m_read):
tr = self.read_pipe_transport()
+ tr.pause_reading()
tr.resume_reading()
self.loop.assert_reader(5, tr._read_ready)
self.assertIsNone(tr._protocol)
self.assertIsNone(tr._loop)
+ def test_pause_reading_on_closed_pipe(self):
+ tr = self.read_pipe_transport()
+ tr.close()
+ test_utils.run_briefly(self.loop)
+ self.assertIsNone(tr._loop)
+ tr.pause_reading()
+
+ def test_pause_reading_on_paused_pipe(self):
+ tr = self.read_pipe_transport()
+ tr.pause_reading()
+ # the second call should do nothing
+ tr.pause_reading()
+
+ def test_resume_reading_on_closed_pipe(self):
+ tr = self.read_pipe_transport()
+ tr.close()
+ test_utils.run_briefly(self.loop)
+ self.assertIsNone(tr._loop)
+ tr.resume_reading()
+
+ def test_resume_reading_on_paused_pipe(self):
+ tr = self.read_pipe_transport()
+ # the pipe is not paused
+ # resuming should do nothing
+ tr.resume_reading()
+
class UnixWritePipeTransportTests(test_utils.TestCase):