<s> """<STR_LIT>""" <EOL> from zope . interface import implements <EOL> from twisted . internet import interfaces <EOL> class BasicProducerConsumerProxy : <EOL> """<STR_LIT>""" <EOL> implements ( interfaces . IProducer , interfaces . IConsumer ) <EOL> consumer = None <EOL> producer = None <EOL> producerIsStreaming = None <EOL> iAmStreaming = True <EOL> outstandingPull = False <EOL> paused = False <EOL> stopped = False <EOL> def __init__ ( self , consumer ) : <EOL> self . _buffer = [ ] <EOL> if consumer is not None : <EOL> self . consumer = consumer <EOL> consumer . registerProducer ( self , self . iAmStreaming ) <EOL> def pauseProducing ( self ) : <EOL> self . paused = True <EOL> if self . producer : <EOL> self . producer . pauseProducing ( ) <EOL> def resumeProducing ( self ) : <EOL> self . paused = False <EOL> if self . _buffer : <EOL> self . consumer . write ( '<STR_LIT>' . join ( self . _buffer ) ) <EOL> self . _buffer [ : ] = [ ] <EOL> else : <EOL> if not self . iAmStreaming : <EOL> self . outstandingPull = True <EOL> if self . producer is not None : <EOL> self . producer . resumeProducing ( ) <EOL> def stopProducing ( self ) : <EOL> if self . producer is not None : <EOL> self . producer . stopProducing ( ) <EOL> if self . consumer is not None : <EOL> del self . consumer <EOL> def write ( self , data ) : <EOL> if self . paused or ( not self . iAmStreaming and not self . outstandingPull ) : <EOL> self . _buffer . append ( data ) <EOL> elif self . consumer is not None : <EOL> self . consumer . write ( data ) <EOL> self . outstandingPull = False <EOL> def finish ( self ) : <EOL> if self . consumer is not None : <EOL> self . consumer . finish ( ) <EOL> self . unregisterProducer ( ) <EOL> def registerProducer ( self , producer , streaming ) : <EOL> self . producer = producer <EOL> self . producerIsStreaming = streaming <EOL> def unregisterProducer ( self ) : <EOL> if self . producer is not None : <EOL> del self . producer <EOL> del self . producerIsStreaming <EOL> if self . consumer : <EOL> self . consumer . unregisterProducer ( ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ , id ( self ) , self . consumer ) <EOL> class ProducerConsumerProxy ( BasicProducerConsumerProxy ) : <EOL> """<STR_LIT>""" <EOL> bufferSize = <NUM_LIT:2> ** <NUM_LIT:2> ** <NUM_LIT:2> ** <NUM_LIT:2> <EOL> producerPaused = False <EOL> unregistered = False <EOL> def pauseProducing ( self ) : <EOL> self . paused = True <EOL> def resumeProducing ( self ) : <EOL> self . paused = False <EOL> if self . _buffer : <EOL> data = '<STR_LIT>' . join ( self . _buffer ) <EOL> bytesSent = self . _writeSomeData ( data ) <EOL> if bytesSent < len ( data ) : <EOL> unsent = data [ bytesSent : ] <EOL> assert not self . iAmStreaming , ( <EOL> "<STR_LIT>" ) <EOL> self . _buffer [ : ] = [ unsent ] <EOL> else : <EOL> self . _buffer [ : ] = [ ] <EOL> else : <EOL> bytesSent = <NUM_LIT:0> <EOL> if ( self . unregistered and bytesSent and not self . _buffer and <EOL> self . consumer is not None ) : <EOL> self . consumer . unregisterProducer ( ) <EOL> if not self . iAmStreaming : <EOL> self . outstandingPull = not bytesSent <EOL> if self . producer is not None : <EOL> bytesBuffered = sum ( [ len ( s ) for s in self . _buffer ] ) <EOL> if self . producerPaused and ( bytesBuffered < self . bufferSize ) : <EOL> self . producerPaused = False <EOL> self . producer . resumeProducing ( ) <EOL> elif self . outstandingPull : <EOL> self . producer . resumeProducing ( ) <EOL> def write ( self , data ) : <EOL> if self . paused or ( not self . iAmStreaming and not self . outstandingPull ) : <EOL> self . _buffer . append ( data ) <EOL> elif self . consumer is not None : <EOL> assert not self . _buffer , ( <EOL> "<STR_LIT>" ) <EOL> bytesSent = self . _writeSomeData ( data ) <EOL> self . outstandingPull = False <EOL> if not bytesSent == len ( data ) : <EOL> assert not self . iAmStreaming , ( <EOL> "<STR_LIT>" ) <EOL> self . _buffer . append ( data [ bytesSent : ] ) <EOL> if ( self . producer is not None ) and self . producerIsStreaming : <EOL> bytesBuffered = sum ( [ len ( s ) for s in self . _buffer ] ) <EOL> if bytesBuffered >= self . bufferSize : <EOL> self . producer . pauseProducing ( ) <EOL> self . producerPaused = True <EOL> def registerProducer ( self , producer , streaming ) : <EOL> self . unregistered = False <EOL> BasicProducerConsumerProxy . registerProducer ( self , producer , streaming ) <EOL> if not streaming : <EOL> producer . resumeProducing ( ) <EOL> def unregisterProducer ( self ) : <EOL> if self . producer is not None : <EOL> del self . producer <EOL> del self . producerIsStreaming <EOL> self . unregistered = True <EOL> if self . consumer and not self . _buffer : <EOL> self . consumer . unregisterProducer ( ) <EOL> def _writeSomeData ( self , data ) : <EOL> """<STR_LIT>""" <EOL> if self . consumer is None : <EOL> return <NUM_LIT:0> <EOL> self . consumer . write ( data ) <EOL> return len ( data ) </s>
