<s> """<STR_LIT>""" <EOL> from __future__ import division , absolute_import <EOL> import socket <EOL> from zope . interface import implementer , classImplements <EOL> import sys <EOL> import warnings <EOL> from heapq import heappush , heappop , heapify <EOL> import traceback <EOL> from twisted . internet . interfaces import IReactorCore , IReactorTime , IReactorThreads <EOL> from twisted . internet . interfaces import IResolverSimple , IReactorPluggableResolver <EOL> from twisted . internet . interfaces import IConnector , IDelayedCall <EOL> from twisted . internet import fdesc , main , error , abstract , defer , threads <EOL> from twisted . python import log , failure , reflect <EOL> from twisted . python . compat import unicode , iteritems <EOL> from twisted . python . runtime import seconds as runtimeSeconds , platform <EOL> from twisted . internet . defer import Deferred , DeferredList <EOL> from twisted . python import threadable <EOL> @ implementer ( IDelayedCall ) <EOL> class DelayedCall : <EOL> debug = False <EOL> _str = None <EOL> def __init__ ( self , time , func , args , kw , cancel , reset , <EOL> seconds = runtimeSeconds ) : <EOL> """<STR_LIT>""" <EOL> self . time , self . func , self . args , self . kw = time , func , args , kw <EOL> self . resetter = reset <EOL> self . canceller = cancel <EOL> self . seconds = seconds <EOL> self . cancelled = self . called = <NUM_LIT:0> <EOL> self . delayed_time = <NUM_LIT:0> <EOL> if self . debug : <EOL> self . creator = traceback . format_stack ( ) [ : - <NUM_LIT:2> ] <EOL> def getTime ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . time + self . delayed_time <EOL> def cancel ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . cancelled : <EOL> raise error . AlreadyCancelled <EOL> elif self . called : <EOL> raise error . AlreadyCalled <EOL> else : <EOL> self . canceller ( self ) <EOL> self . cancelled = <NUM_LIT:1> <EOL> if self . debug : <EOL> self . _str = bytes ( self ) <EOL> del self . func , self . args , self . kw <EOL> def reset ( self , secondsFromNow ) : <EOL> """<STR_LIT>""" <EOL> if self . cancelled : <EOL> raise error . AlreadyCancelled <EOL> elif self . called : <EOL> raise error . AlreadyCalled <EOL> else : <EOL> newTime = self . seconds ( ) + secondsFromNow <EOL> if newTime < self . time : <EOL> self . delayed_time = <NUM_LIT:0> <EOL> self . time = newTime <EOL> self . resetter ( self ) <EOL> else : <EOL> self . delayed_time = newTime - self . time <EOL> def delay ( self , secondsLater ) : <EOL> """<STR_LIT>""" <EOL> if self . cancelled : <EOL> raise error . AlreadyCancelled <EOL> elif self . called : <EOL> raise error . AlreadyCalled <EOL> else : <EOL> self . delayed_time += secondsLater <EOL> if self . delayed_time < <NUM_LIT:0> : <EOL> self . activate_delay ( ) <EOL> self . resetter ( self ) <EOL> def activate_delay ( self ) : <EOL> self . time += self . delayed_time <EOL> self . delayed_time = <NUM_LIT:0> <EOL> def active ( self ) : <EOL> """<STR_LIT>""" <EOL> return not ( self . cancelled or self . called ) <EOL> def __le__ ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return self . time <= other . time <EOL> def __lt__ ( self , other ) : <EOL> """<STR_LIT>""" <EOL> return self . time < other . time <EOL> def __str__ ( self ) : <EOL> if self . _str is not None : <EOL> return self . _str <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> if hasattr ( self . func , '<STR_LIT>' ) : <EOL> func = self . func . __qualname__ <EOL> elif hasattr ( self . func , '<STR_LIT>' ) : <EOL> func = self . func . func_name <EOL> if hasattr ( self . func , '<STR_LIT>' ) : <EOL> func = self . func . im_class . __name__ + '<STR_LIT:.>' + func <EOL> else : <EOL> func = reflect . safe_repr ( self . func ) <EOL> else : <EOL> func = None <EOL> now = self . seconds ( ) <EOL> L = [ "<STR_LIT>" % ( <EOL> id ( self ) , self . time - now , self . called , <EOL> self . cancelled ) ] <EOL> if func is not None : <EOL> L . extend ( ( "<STR_LIT:U+0020>" , func , "<STR_LIT:(>" ) ) <EOL> if self . args : <EOL> L . append ( "<STR_LIT:U+002CU+0020>" . join ( [ reflect . safe_repr ( e ) for e in self . args ] ) ) <EOL> if self . kw : <EOL> L . append ( "<STR_LIT:U+002CU+0020>" ) <EOL> if self . kw : <EOL> L . append ( "<STR_LIT:U+002CU+0020>" . join ( [ '<STR_LIT>' % ( k , reflect . safe_repr ( v ) ) for ( k , v ) in self . kw . items ( ) ] ) ) <EOL> L . append ( "<STR_LIT:)>" ) <EOL> if self . debug : <EOL> L . append ( "<STR_LIT>" % ( '<STR_LIT:U+0020>' . join ( self . creator ) ) ) <EOL> L . append ( '<STR_LIT:>>' ) <EOL> return "<STR_LIT>" . join ( L ) <EOL> @ implementer ( IResolverSimple ) <EOL> class ThreadedResolver ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , reactor ) : <EOL> self . reactor = reactor <EOL> self . _runningQueries = { } <EOL> def _fail ( self , name , err ) : <EOL> err = error . DNSLookupError ( "<STR_LIT>" % ( name , err ) ) <EOL> return failure . Failure ( err ) <EOL> def _cleanup ( self , name , lookupDeferred ) : <EOL> userDeferred , cancelCall = self . _runningQueries [ lookupDeferred ] <EOL> del self . _runningQueries [ lookupDeferred ] <EOL> userDeferred . errback ( self . _fail ( name , "<STR_LIT>" ) ) <EOL> def _checkTimeout ( self , result , name , lookupDeferred ) : <EOL> try : <EOL> userDeferred , cancelCall = self . _runningQueries [ lookupDeferred ] <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> del self . _runningQueries [ lookupDeferred ] <EOL> cancelCall . cancel ( ) <EOL> if isinstance ( result , failure . Failure ) : <EOL> userDeferred . errback ( self . _fail ( name , result . getErrorMessage ( ) ) ) <EOL> else : <EOL> userDeferred . callback ( result ) <EOL> def getHostByName ( self , name , timeout = ( <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:11> , <NUM_LIT> ) ) : <EOL> """<STR_LIT>""" <EOL> if timeout : <EOL> timeoutDelay = sum ( timeout ) <EOL> else : <EOL> timeoutDelay = <NUM_LIT> <EOL> userDeferred = defer . Deferred ( ) <EOL> lookupDeferred = threads . deferToThreadPool ( <EOL> self . reactor , self . reactor . getThreadPool ( ) , <EOL> socket . gethostbyname , name ) <EOL> cancelCall = self . reactor . callLater ( <EOL> timeoutDelay , self . _cleanup , name , lookupDeferred ) <EOL> self . _runningQueries [ lookupDeferred ] = ( userDeferred , cancelCall ) <EOL> lookupDeferred . addBoth ( self . _checkTimeout , name , lookupDeferred ) <EOL> return userDeferred <EOL> @ implementer ( IResolverSimple ) <EOL> class BlockingResolver : <EOL> def getHostByName ( self , name , timeout = ( <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:11> , <NUM_LIT> ) ) : <EOL> try : <EOL> address = socket . gethostbyname ( name ) <EOL> except socket . error : <EOL> msg = "<STR_LIT>" % ( name , ) <EOL> err = error . DNSLookupError ( msg ) <EOL> return defer . fail ( err ) <EOL> else : <EOL> return defer . succeed ( address ) <EOL> class _ThreePhaseEvent ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . before = [ ] <EOL> self . during = [ ] <EOL> self . after = [ ] <EOL> self . state = '<STR_LIT>' <EOL> def addTrigger ( self , phase , callable , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if phase not in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> raise KeyError ( "<STR_LIT>" ) <EOL> getattr ( self , phase ) . append ( ( callable , args , kwargs ) ) <EOL> return phase , callable , args , kwargs <EOL> def removeTrigger ( self , handle ) : <EOL> """<STR_LIT>""" <EOL> return getattr ( self , '<STR_LIT>' + self . state ) ( handle ) <EOL> def removeTrigger_BASE ( self , handle ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> phase , callable , args , kwargs = handle <EOL> except ( TypeError , ValueError ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> else : <EOL> if phase not in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> raise KeyError ( "<STR_LIT>" ) <EOL> getattr ( self , phase ) . remove ( ( callable , args , kwargs ) ) <EOL> def removeTrigger_BEFORE ( self , handle ) : <EOL> """<STR_LIT>""" <EOL> phase , callable , args , kwargs = handle <EOL> if phase != '<STR_LIT>' : <EOL> return self . removeTrigger_BASE ( handle ) <EOL> if ( callable , args , kwargs ) in self . finishedBefore : <EOL> warnings . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> category = DeprecationWarning , <EOL> stacklevel = <NUM_LIT:3> ) <EOL> else : <EOL> self . removeTrigger_BASE ( handle ) <EOL> def fireEvent ( self ) : <EOL> """<STR_LIT>""" <EOL> self . state = '<STR_LIT>' <EOL> self . finishedBefore = [ ] <EOL> beforeResults = [ ] <EOL> while self . before : <EOL> callable , args , kwargs = self . before . pop ( <NUM_LIT:0> ) <EOL> self . finishedBefore . append ( ( callable , args , kwargs ) ) <EOL> try : <EOL> result = callable ( * args , ** kwargs ) <EOL> except : <EOL> log . err ( ) <EOL> else : <EOL> if isinstance ( result , Deferred ) : <EOL> beforeResults . append ( result ) <EOL> DeferredList ( beforeResults ) . addCallback ( self . _continueFiring ) <EOL> def _continueFiring ( self , ignored ) : <EOL> """<STR_LIT>""" <EOL> self . state = '<STR_LIT>' <EOL> self . finishedBefore = [ ] <EOL> for phase in self . during , self . after : <EOL> while phase : <EOL> callable , args , kwargs = phase . pop ( <NUM_LIT:0> ) <EOL> try : <EOL> callable ( * args , ** kwargs ) <EOL> except : <EOL> log . err ( ) <EOL> @ implementer ( IReactorCore , IReactorTime , IReactorPluggableResolver ) <EOL> class ReactorBase ( object ) : <EOL> """<STR_LIT>""" <EOL> _registerAsIOThread = True <EOL> _stopped = True <EOL> installed = False <EOL> usingThreads = False <EOL> resolver = BlockingResolver ( ) <EOL> __name__ = "<STR_LIT>" <EOL> def __init__ ( self ) : <EOL> self . threadCallQueue = [ ] <EOL> self . _eventTriggers = { } <EOL> self . _pendingTimedCalls = [ ] <EOL> self . _newTimedCalls = [ ] <EOL> self . _cancellations = <NUM_LIT:0> <EOL> self . running = False <EOL> self . _started = False <EOL> self . _justStopped = False <EOL> self . _startedBefore = False <EOL> self . _internalReaders = set ( ) <EOL> self . waker = None <EOL> self . addSystemEventTrigger ( <EOL> '<STR_LIT>' , '<STR_LIT>' , self . _reallyStartRunning ) <EOL> self . addSystemEventTrigger ( '<STR_LIT>' , '<STR_LIT>' , self . crash ) <EOL> self . addSystemEventTrigger ( '<STR_LIT>' , '<STR_LIT>' , self . disconnectAll ) <EOL> if platform . supportsThreads ( ) : <EOL> self . _initThreads ( ) <EOL> self . installWaker ( ) <EOL> _lock = None <EOL> def installWaker ( self ) : <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" ) <EOL> def installResolver ( self , resolver ) : <EOL> assert IResolverSimple . providedBy ( resolver ) <EOL> oldResolver = self . resolver <EOL> self . resolver = resolver <EOL> return oldResolver <EOL> def wakeUp ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . waker : <EOL> self . waker . wakeUp ( ) <EOL> def doIteration ( self , delay ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" ) <EOL> def addReader ( self , reader ) : <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" ) <EOL> def addWriter ( self , writer ) : <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" ) <EOL> def removeReader ( self , reader ) : <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" ) <EOL> def removeWriter ( self , writer ) : <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" ) <EOL> def removeAll ( self ) : <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" ) <EOL> def getReaders ( self ) : <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" ) <EOL> def getWriters ( self ) : <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" ) <EOL> def resolve ( self , name , timeout = ( <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:11> , <NUM_LIT> ) ) : <EOL> """<STR_LIT>""" <EOL> if not name : <EOL> return defer . succeed ( '<STR_LIT>' ) <EOL> if abstract . isIPAddress ( name ) : <EOL> return defer . succeed ( name ) <EOL> return self . resolver . getHostByName ( name , timeout ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _stopped : <EOL> raise error . ReactorNotRunning ( <EOL> "<STR_LIT>" ) <EOL> self . _stopped = True <EOL> self . _justStopped = True <EOL> self . _startedBefore = True <EOL> def crash ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _started = False <EOL> self . running = False <EOL> self . addSystemEventTrigger ( <EOL> '<STR_LIT>' , '<STR_LIT>' , self . _reallyStartRunning ) <EOL> def sigInt ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . callFromThread ( self . stop ) <EOL> def sigBreak ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . callFromThread ( self . stop ) <EOL> def sigTerm ( self , * args ) : <EOL> """<STR_LIT>""" <EOL> log . msg ( "<STR_LIT>" ) <EOL> self . callFromThread ( self . stop ) <EOL> def disconnectAll ( self ) : <EOL> """<STR_LIT>""" <EOL> selectables = self . removeAll ( ) <EOL> for reader in selectables : <EOL> log . callWithLogger ( reader , <EOL> reader . connectionLost , <EOL> failure . Failure ( main . CONNECTION_LOST ) ) <EOL> def iterate ( self , delay = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> self . runUntilCurrent ( ) <EOL> self . doIteration ( delay ) <EOL> def fireSystemEvent ( self , eventType ) : <EOL> """<STR_LIT>""" <EOL> event = self . _eventTriggers . get ( eventType ) <EOL> if event is not None : <EOL> event . fireEvent ( ) <EOL> def addSystemEventTrigger ( self , _phase , _eventType , _f , * args , ** kw ) : <EOL> """<STR_LIT>""" <EOL> assert callable ( _f ) , "<STR_LIT>" % _f <EOL> if _eventType not in self . _eventTriggers : <EOL> self . _eventTriggers [ _eventType ] = _ThreePhaseEvent ( ) <EOL> return ( _eventType , self . _eventTriggers [ _eventType ] . addTrigger ( <EOL> _phase , _f , * args , ** kw ) ) <EOL> def removeSystemEventTrigger ( self , triggerID ) : <EOL> """<STR_LIT>""" <EOL> eventType , handle = triggerID <EOL> self . _eventTriggers [ eventType ] . removeTrigger ( handle ) <EOL> def callWhenRunning ( self , _callable , * args , ** kw ) : <EOL> """<STR_LIT>""" <EOL> if self . running : <EOL> _callable ( * args , ** kw ) <EOL> else : <EOL> return self . addSystemEventTrigger ( '<STR_LIT>' , '<STR_LIT>' , <EOL> _callable , * args , ** kw ) <EOL> def startRunning ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _started : <EOL> raise error . ReactorAlreadyRunning ( ) <EOL> if self . _startedBefore : <EOL> raise error . ReactorNotRestartable ( ) <EOL> self . _started = True <EOL> self . _stopped = False <EOL> if self . _registerAsIOThread : <EOL> threadable . registerAsIOThread ( ) <EOL> self . fireSystemEvent ( '<STR_LIT>' ) <EOL> def _reallyStartRunning ( self ) : <EOL> """<STR_LIT>""" <EOL> self . running = True <EOL> seconds = staticmethod ( runtimeSeconds ) <EOL> def callLater ( self , _seconds , _f , * args , ** kw ) : <EOL> """<STR_LIT>""" <EOL> assert callable ( _f ) , "<STR_LIT>" % _f <EOL> assert _seconds >= <NUM_LIT:0> , "<STR_LIT>" % ( _seconds , ) <EOL> tple = DelayedCall ( self . seconds ( ) + _seconds , _f , args , kw , <EOL> self . _cancelCallLater , <EOL> self . _moveCallLaterSooner , <EOL> seconds = self . seconds ) <EOL> self . _newTimedCalls . append ( tple ) <EOL> return tple <EOL> def _moveCallLaterSooner ( self , tple ) : <EOL> heap = self . _pendingTimedCalls <EOL> try : <EOL> pos = heap . index ( tple ) <EOL> elt = heap [ pos ] <EOL> while pos != <NUM_LIT:0> : <EOL> parent = ( pos - <NUM_LIT:1> ) // <NUM_LIT:2> <EOL> if heap [ parent ] <= elt : <EOL> break <EOL> heap [ pos ] = heap [ parent ] <EOL> pos = parent <EOL> heap [ pos ] = elt <EOL> except ValueError : <EOL> pass <EOL> def _cancelCallLater ( self , tple ) : <EOL> self . _cancellations += <NUM_LIT:1> <EOL> def getDelayedCalls ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ x for x in ( self . _pendingTimedCalls + self . _newTimedCalls ) if not x . cancelled ] <EOL> def _insertNewDelayedCalls ( self ) : <EOL> for call in self . _newTimedCalls : <EOL> if call . cancelled : <EOL> self . _cancellations -= <NUM_LIT:1> <EOL> else : <EOL> call . activate_delay ( ) <EOL> heappush ( self . _pendingTimedCalls , call ) <EOL> self . _newTimedCalls = [ ] <EOL> def timeout ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _insertNewDelayedCalls ( ) <EOL> if not self . _pendingTimedCalls : <EOL> return None <EOL> delay = self . _pendingTimedCalls [ <NUM_LIT:0> ] . time - self . seconds ( ) <EOL> longest = <NUM_LIT> <EOL> return max ( <NUM_LIT:0> , min ( longest , delay ) ) <EOL> def runUntilCurrent ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . threadCallQueue : <EOL> count = <NUM_LIT:0> <EOL> total = len ( self . threadCallQueue ) <EOL> for ( f , a , kw ) in self . threadCallQueue : <EOL> try : <EOL> f ( * a , ** kw ) <EOL> except : <EOL> log . err ( ) <EOL> count += <NUM_LIT:1> <EOL> if count == total : <EOL> break <EOL> del self . threadCallQueue [ : count ] <EOL> if self . threadCallQueue : <EOL> self . wakeUp ( ) <EOL> self . _insertNewDelayedCalls ( ) <EOL> now = self . seconds ( ) <EOL> while self . _pendingTimedCalls and ( self . _pendingTimedCalls [ <NUM_LIT:0> ] . time <= now ) : <EOL> call = heappop ( self . _pendingTimedCalls ) <EOL> if call . cancelled : <EOL> self . _cancellations -= <NUM_LIT:1> <EOL> continue <EOL> if call . delayed_time > <NUM_LIT:0> : <EOL> call . activate_delay ( ) <EOL> heappush ( self . _pendingTimedCalls , call ) <EOL> continue <EOL> try : <EOL> call . called = <NUM_LIT:1> <EOL> call . func ( * call . args , ** call . kw ) <EOL> except : <EOL> log . deferr ( ) <EOL> if hasattr ( call , "<STR_LIT>" ) : <EOL> e = "<STR_LIT:\n>" <EOL> e += "<STR_LIT>" + "<STR_LIT>" <EOL> e += "<STR_LIT>" <EOL> e += "<STR_LIT>" . join ( call . creator ) . rstrip ( ) . replace ( "<STR_LIT:\n>" , "<STR_LIT>" ) <EOL> e += "<STR_LIT:\n>" <EOL> log . msg ( e ) <EOL> if ( self . _cancellations > <NUM_LIT:50> and <EOL> self . _cancellations > len ( self . _pendingTimedCalls ) >> <NUM_LIT:1> ) : <EOL> self . _cancellations = <NUM_LIT:0> <EOL> self . _pendingTimedCalls = [ x for x in self . _pendingTimedCalls <EOL> if not x . cancelled ] <EOL> heapify ( self . _pendingTimedCalls ) <EOL> if self . _justStopped : <EOL> self . _justStopped = False <EOL> self . fireSystemEvent ( "<STR_LIT>" ) <EOL> def _checkProcessArgs ( self , args , env ) : <EOL> """<STR_LIT>""" <EOL> defaultEncoding = sys . getdefaultencoding ( ) <EOL> def argChecker ( arg ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( arg , unicode ) : <EOL> try : <EOL> arg = arg . encode ( defaultEncoding ) <EOL> except UnicodeEncodeError : <EOL> return None <EOL> warnings . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> category = DeprecationWarning , <EOL> stacklevel = <NUM_LIT:4> ) <EOL> if isinstance ( arg , bytes ) and b'<STR_LIT>' not in arg : <EOL> return arg <EOL> return None <EOL> if not isinstance ( args , ( tuple , list ) ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> outputArgs = [ ] <EOL> for arg in args : <EOL> arg = argChecker ( arg ) <EOL> if arg is None : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> else : <EOL> outputArgs . append ( arg ) <EOL> outputEnv = None <EOL> if env is not None : <EOL> outputEnv = { } <EOL> for key , val in iteritems ( env ) : <EOL> key = argChecker ( key ) <EOL> if key is None : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> val = argChecker ( val ) <EOL> if val is None : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> outputEnv [ key ] = val <EOL> return outputArgs , outputEnv <EOL> if platform . supportsThreads ( ) : <EOL> threadpool = None <EOL> _threadpoolStartupID = None <EOL> threadpoolShutdownID = None <EOL> def _initThreads ( self ) : <EOL> self . usingThreads = True <EOL> self . resolver = ThreadedResolver ( self ) <EOL> def callFromThread ( self , f , * args , ** kw ) : <EOL> """<STR_LIT>""" <EOL> assert callable ( f ) , "<STR_LIT>" % ( f , ) <EOL> self . threadCallQueue . append ( ( f , args , kw ) ) <EOL> self . wakeUp ( ) <EOL> def _initThreadPool ( self ) : <EOL> """<STR_LIT>""" <EOL> from twisted . python import threadpool <EOL> self . threadpool = threadpool . ThreadPool ( <EOL> <NUM_LIT:0> , <NUM_LIT:10> , '<STR_LIT>' ) <EOL> self . _threadpoolStartupID = self . callWhenRunning ( <EOL> self . threadpool . start ) <EOL> self . threadpoolShutdownID = self . addSystemEventTrigger ( <EOL> '<STR_LIT>' , '<STR_LIT>' , self . _stopThreadPool ) <EOL> def _uninstallHandler ( self ) : <EOL> pass <EOL> def _stopThreadPool ( self ) : <EOL> """<STR_LIT>""" <EOL> triggers = [ self . _threadpoolStartupID , self . threadpoolShutdownID ] <EOL> for trigger in filter ( None , triggers ) : <EOL> try : <EOL> self . removeSystemEventTrigger ( trigger ) <EOL> except ValueError : <EOL> pass <EOL> self . _threadpoolStartupID = None <EOL> self . threadpoolShutdownID = None <EOL> self . threadpool . stop ( ) <EOL> self . threadpool = None <EOL> def getThreadPool ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . threadpool is None : <EOL> self . _initThreadPool ( ) <EOL> return self . threadpool <EOL> def callInThread ( self , _callable , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . getThreadPool ( ) . callInThread ( _callable , * args , ** kwargs ) <EOL> def suggestThreadPoolSize ( self , size ) : <EOL> """<STR_LIT>""" <EOL> self . getThreadPool ( ) . adjustPoolsize ( maxthreads = size ) <EOL> else : <EOL> def callFromThread ( self , f , * args , ** kw ) : <EOL> assert callable ( f ) , "<STR_LIT>" % ( f , ) <EOL> self . threadCallQueue . append ( ( f , args , kw ) ) <EOL> if platform . supportsThreads ( ) : <EOL> classImplements ( ReactorBase , IReactorThreads ) <EOL> @ implementer ( IConnector ) <EOL> class BaseConnector : <EOL> """<STR_LIT>""" <EOL> timeoutID = None <EOL> factoryStarted = <NUM_LIT:0> <EOL> def __init__ ( self , factory , timeout , reactor ) : <EOL> self . state = "<STR_LIT>" <EOL> self . reactor = reactor <EOL> self . factory = factory <EOL> self . timeout = timeout <EOL> def disconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . state == '<STR_LIT>' : <EOL> self . stopConnecting ( ) <EOL> elif self . state == '<STR_LIT>' : <EOL> self . transport . loseConnection ( ) <EOL> def connect ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . state != "<STR_LIT>" : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> self . state = "<STR_LIT>" <EOL> if not self . factoryStarted : <EOL> self . factory . doStart ( ) <EOL> self . factoryStarted = <NUM_LIT:1> <EOL> self . transport = transport = self . _makeTransport ( ) <EOL> if self . timeout is not None : <EOL> self . timeoutID = self . reactor . callLater ( self . timeout , transport . failIfNotConnected , error . TimeoutError ( ) ) <EOL> self . factory . startedConnecting ( self ) <EOL> def stopConnecting ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . state != "<STR_LIT>" : <EOL> raise error . NotConnectingError ( "<STR_LIT>" ) <EOL> self . state = "<STR_LIT>" <EOL> self . transport . failIfNotConnected ( error . UserError ( ) ) <EOL> del self . transport <EOL> def cancelTimeout ( self ) : <EOL> if self . timeoutID is not None : <EOL> try : <EOL> self . timeoutID . cancel ( ) <EOL> except ValueError : <EOL> pass <EOL> del self . timeoutID <EOL> def buildProtocol ( self , addr ) : <EOL> self . state = "<STR_LIT>" <EOL> self . cancelTimeout ( ) <EOL> return self . factory . buildProtocol ( addr ) <EOL> def connectionFailed ( self , reason ) : <EOL> self . cancelTimeout ( ) <EOL> self . transport = None <EOL> self . state = "<STR_LIT>" <EOL> self . factory . clientConnectionFailed ( self , reason ) <EOL> if self . state == "<STR_LIT>" : <EOL> self . factory . doStop ( ) <EOL> self . factoryStarted = <NUM_LIT:0> <EOL> def connectionLost ( self , reason ) : <EOL> self . state = "<STR_LIT>" <EOL> self . factory . clientConnectionLost ( self , reason ) <EOL> if self . state == "<STR_LIT>" : <EOL> self . factory . doStop ( ) <EOL> self . factoryStarted = <NUM_LIT:0> <EOL> def getDestination ( self ) : <EOL> raise NotImplementedError ( <EOL> reflect . qual ( self . __class__ ) + "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class BasePort ( abstract . FileDescriptor ) : <EOL> """<STR_LIT>""" <EOL> addressFamily = None <EOL> socketType = None <EOL> def createInternetSocket ( self ) : <EOL> s = socket . socket ( self . addressFamily , self . socketType ) <EOL> s . setblocking ( <NUM_LIT:0> ) <EOL> fdesc . _setCloseOnExec ( s . fileno ( ) ) <EOL> return s <EOL> def doWrite ( self ) : <EOL> """<STR_LIT>""" <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" % reflect . qual ( self . __class__ ) ) <EOL> class _SignalReactorMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> _installSignalHandlers = False <EOL> def _handleSignals ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> import signal <EOL> except ImportError : <EOL> log . msg ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return <EOL> if signal . getsignal ( signal . SIGINT ) == signal . default_int_handler : <EOL> signal . signal ( signal . SIGINT , self . sigInt ) <EOL> signal . signal ( signal . SIGTERM , self . sigTerm ) <EOL> if hasattr ( signal , "<STR_LIT>" ) : <EOL> signal . signal ( signal . SIGBREAK , self . sigBreak ) <EOL> def startRunning ( self , installSignalHandlers = True ) : <EOL> """<STR_LIT>""" <EOL> self . _installSignalHandlers = installSignalHandlers <EOL> ReactorBase . startRunning ( self ) <EOL> def _reallyStartRunning ( self ) : <EOL> """<STR_LIT>""" <EOL> ReactorBase . _reallyStartRunning ( self ) <EOL> if self . _installSignalHandlers : <EOL> self . _handleSignals ( ) <EOL> def run ( self , installSignalHandlers = True ) : <EOL> self . startRunning ( installSignalHandlers = installSignalHandlers ) <EOL> self . mainLoop ( ) <EOL> def mainLoop ( self ) : <EOL> while self . _started : <EOL> try : <EOL> while self . _started : <EOL> self . runUntilCurrent ( ) <EOL> t2 = self . timeout ( ) <EOL> t = self . running and t2 <EOL> self . doIteration ( t ) <EOL> except : <EOL> log . msg ( "<STR_LIT>" ) <EOL> log . err ( ) <EOL> else : <EOL> log . msg ( '<STR_LIT>' ) <EOL> __all__ = [ ] </s>
