<s> """<STR_LIT>""" <EOL> from zope . interface import implementer <EOL> from twisted . internet . interfaces import IConsumer , IPushProducer <EOL> MIN_TIMEOUT = <NUM_LIT> <EOL> MAX_TIMEOUT = <NUM_LIT:0.1> <EOL> class _PollableResource : <EOL> active = True <EOL> def activate ( self ) : <EOL> self . active = True <EOL> def deactivate ( self ) : <EOL> self . active = False <EOL> class _PollingTimer : <EOL> def __init__ ( self , reactor ) : <EOL> self . reactor = reactor <EOL> self . _resources = [ ] <EOL> self . _pollTimer = None <EOL> self . _currentTimeout = MAX_TIMEOUT <EOL> self . _paused = False <EOL> def _addPollableResource ( self , res ) : <EOL> self . _resources . append ( res ) <EOL> self . _checkPollingState ( ) <EOL> def _checkPollingState ( self ) : <EOL> for resource in self . _resources : <EOL> if resource . active : <EOL> self . _startPolling ( ) <EOL> break <EOL> else : <EOL> self . _stopPolling ( ) <EOL> def _startPolling ( self ) : <EOL> if self . _pollTimer is None : <EOL> self . _pollTimer = self . _reschedule ( ) <EOL> def _stopPolling ( self ) : <EOL> if self . _pollTimer is not None : <EOL> self . _pollTimer . cancel ( ) <EOL> self . _pollTimer = None <EOL> def _pause ( self ) : <EOL> self . _paused = True <EOL> def _unpause ( self ) : <EOL> self . _paused = False <EOL> self . _checkPollingState ( ) <EOL> def _reschedule ( self ) : <EOL> if not self . _paused : <EOL> return self . reactor . callLater ( self . _currentTimeout , self . _pollEvent ) <EOL> def _pollEvent ( self ) : <EOL> workUnits = <NUM_LIT:0.> <EOL> anyActive = [ ] <EOL> for resource in self . _resources : <EOL> if resource . active : <EOL> workUnits += resource . checkWork ( ) <EOL> if resource . active : <EOL> anyActive . append ( resource ) <EOL> newTimeout = self . _currentTimeout <EOL> if workUnits : <EOL> newTimeout = self . _currentTimeout / ( workUnits + <NUM_LIT:1.> ) <EOL> if newTimeout < MIN_TIMEOUT : <EOL> newTimeout = MIN_TIMEOUT <EOL> else : <EOL> newTimeout = self . _currentTimeout * <NUM_LIT> <EOL> if newTimeout > MAX_TIMEOUT : <EOL> newTimeout = MAX_TIMEOUT <EOL> self . _currentTimeout = newTimeout <EOL> if anyActive : <EOL> self . _pollTimer = self . _reschedule ( ) <EOL> import win32pipe <EOL> import win32file <EOL> import win32api <EOL> import pywintypes <EOL> @ implementer ( IPushProducer ) <EOL> class _PollableReadPipe ( _PollableResource ) : <EOL> def __init__ ( self , pipe , receivedCallback , lostCallback ) : <EOL> self . pipe = pipe <EOL> self . receivedCallback = receivedCallback <EOL> self . lostCallback = lostCallback <EOL> def checkWork ( self ) : <EOL> finished = <NUM_LIT:0> <EOL> fullDataRead = [ ] <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> buffer , bytesToRead , result = win32pipe . PeekNamedPipe ( self . pipe , <NUM_LIT:1> ) <EOL> if not bytesToRead : <EOL> break <EOL> hr , data = win32file . ReadFile ( self . pipe , bytesToRead , None ) <EOL> fullDataRead . append ( data ) <EOL> except win32api . error : <EOL> finished = <NUM_LIT:1> <EOL> break <EOL> dataBuf = '<STR_LIT>' . join ( fullDataRead ) <EOL> if dataBuf : <EOL> self . receivedCallback ( dataBuf ) <EOL> if finished : <EOL> self . cleanup ( ) <EOL> return len ( dataBuf ) <EOL> def cleanup ( self ) : <EOL> self . deactivate ( ) <EOL> self . lostCallback ( ) <EOL> def close ( self ) : <EOL> try : <EOL> win32api . CloseHandle ( self . pipe ) <EOL> except pywintypes . error : <EOL> pass <EOL> def stopProducing ( self ) : <EOL> self . close ( ) <EOL> def pauseProducing ( self ) : <EOL> self . deactivate ( ) <EOL> def resumeProducing ( self ) : <EOL> self . activate ( ) <EOL> FULL_BUFFER_SIZE = <NUM_LIT:64> * <NUM_LIT> <EOL> @ implementer ( IConsumer ) <EOL> class _PollableWritePipe ( _PollableResource ) : <EOL> def __init__ ( self , writePipe , lostCallback ) : <EOL> self . disconnecting = False <EOL> self . producer = None <EOL> self . producerPaused = False <EOL> self . streamingProducer = <NUM_LIT:0> <EOL> self . outQueue = [ ] <EOL> self . writePipe = writePipe <EOL> self . lostCallback = lostCallback <EOL> try : <EOL> win32pipe . SetNamedPipeHandleState ( writePipe , <EOL> win32pipe . PIPE_NOWAIT , <EOL> None , <EOL> None ) <EOL> except pywintypes . error : <EOL> pass <EOL> def close ( self ) : <EOL> self . disconnecting = True <EOL> def bufferFull ( self ) : <EOL> if self . producer is not None : <EOL> self . producerPaused = True <EOL> self . producer . pauseProducing ( ) <EOL> def bufferEmpty ( self ) : <EOL> if self . producer is not None and ( ( not self . streamingProducer ) or <EOL> self . producerPaused ) : <EOL> self . producer . producerPaused = False <EOL> self . producer . resumeProducing ( ) <EOL> return True <EOL> return False <EOL> def registerProducer ( self , producer , streaming ) : <EOL> """<STR_LIT>""" <EOL> if self . producer is not None : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( producer , self . producer ) ) <EOL> if not self . active : <EOL> producer . stopProducing ( ) <EOL> else : <EOL> self . producer = producer <EOL> self . streamingProducer = streaming <EOL> if not streaming : <EOL> producer . resumeProducing ( ) <EOL> def unregisterProducer ( self ) : <EOL> """<STR_LIT>""" <EOL> self . producer = None <EOL> def writeConnectionLost ( self ) : <EOL> self . deactivate ( ) <EOL> try : <EOL> win32api . CloseHandle ( self . writePipe ) <EOL> except pywintypes . error : <EOL> pass <EOL> self . lostCallback ( ) <EOL> def writeSequence ( self , seq ) : <EOL> """<STR_LIT>""" <EOL> if unicode in map ( type , seq ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . outQueue . extend ( seq ) <EOL> def write ( self , data ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( data , unicode ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if self . disconnecting : <EOL> return <EOL> self . outQueue . append ( data ) <EOL> if sum ( map ( len , self . outQueue ) ) > FULL_BUFFER_SIZE : <EOL> self . bufferFull ( ) <EOL> def checkWork ( self ) : <EOL> numBytesWritten = <NUM_LIT:0> <EOL> if not self . outQueue : <EOL> if self . disconnecting : <EOL> self . writeConnectionLost ( ) <EOL> return <NUM_LIT:0> <EOL> try : <EOL> win32file . WriteFile ( self . writePipe , '<STR_LIT>' , None ) <EOL> except pywintypes . error : <EOL> self . writeConnectionLost ( ) <EOL> return numBytesWritten <EOL> while self . outQueue : <EOL> data = self . outQueue . pop ( <NUM_LIT:0> ) <EOL> errCode = <NUM_LIT:0> <EOL> try : <EOL> errCode , nBytesWritten = win32file . WriteFile ( self . writePipe , <EOL> data , None ) <EOL> except win32api . error : <EOL> self . writeConnectionLost ( ) <EOL> break <EOL> else : <EOL> numBytesWritten += nBytesWritten <EOL> if len ( data ) > nBytesWritten : <EOL> self . outQueue . insert ( <NUM_LIT:0> , data [ nBytesWritten : ] ) <EOL> break <EOL> else : <EOL> resumed = self . bufferEmpty ( ) <EOL> if not resumed and self . disconnecting : <EOL> self . writeConnectionLost ( ) <EOL> return numBytesWritten </s>
