1 # -*- test-case-name: twisted.test.test_stdio.StandardInputOutputTestCase.test_producer -*-
2 # Copyright (c) Twisted Matrix Laboratories.
3 # See LICENSE for details.
6 Main program for the child process run by
7 L{twisted.test.test_stdio.StandardInputOutputTestCase.test_producer} to test
8 that process transports implement IProducer properly.
13 from twisted.internet import stdio, protocol
14 from twisted.python import log, reflect
16 class ProducerChild(protocol.Protocol):
20 def connectionLost(self, reason):
21 log.msg("*****OVER*****")
22 reactor.callLater(1, reactor.stop)
26 def dataReceived(self, bytes):
29 log.startLogging(sys.stderr)
30 log.msg("dataReceived while transport paused!")
31 self.transport.loseConnection()
33 self.transport.write(bytes)
34 if self.buf.endswith('\n0\n'):
35 self.transport.loseConnection()
42 self.transport.pauseProducing()
43 reactor.callLater(0.01, self.unpause)
48 self.transport.resumeProducing()
51 if __name__ == '__main__':
52 reflect.namedAny(sys.argv[1]).install()
53 from twisted.internet import reactor
54 stdio.StandardIO(ProducerChild())