twisted_test.py 文件源码

def _testReadWrite(self):
        """
        In this test the writer writes an 'x' to its fd. The reader
        reads it, check the value and ends the test.
        """
        self.shouldWrite = True
        def checkReadInput(fd):
            self.assertEquals(fd.read(), 'x')
            self._reactor.stop()
        def writeOnce(fd):
            if self.shouldWrite:
                self.shouldWrite = False
                fd.write('x')
        self._reader = Reader(self._p1, checkReadInput)
        self._writer = Writer(self._p2, writeOnce)

        self._reactor.addWriter(self._writer)

        # Test that adding the reader twice adds it only once to
        # IOLoop.
        self._reactor.addReader(self._reader)
        self._reactor.addReader(self._reader)