test_loopback.py 文件源码

def test_pullProducer(self):
        """
        Test a pull producer registered against a loopback transport.
        """
        @implementer(IPullProducer)
        class PullProducer(object):
            def __init__(self, toProduce):
                self.toProduce = toProduce

            def start(self, consumer):
                self.consumer = consumer
                self.consumer.registerProducer(self, False)

            def resumeProducing(self):
                self.consumer.write(self.toProduce.pop(0))
                if not self.toProduce:
                    self.consumer.unregisterProducer()
        return self._producertest(PullProducer)