<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import select <EOL> import threading <EOL> import weakref <EOL> import warnings <EOL> from functools import wraps <EOL> import imp <EOL> from twisted . python import threadable <EOL> from twisted . python . runtime import platform <EOL> from twisted . python . failure import Failure <EOL> from twisted . python . log import PythonLoggingObserver , err <EOL> from twisted . internet . defer import maybeDeferred <EOL> from twisted . internet . task import LoopingCall <EOL> from . _util import synchronized <EOL> from . _resultstore import ResultStore <EOL> _store = ResultStore ( ) <EOL> if hasattr ( weakref , "<STR_LIT>" ) : <EOL> WeakSet = weakref . WeakSet <EOL> else : <EOL> class WeakSet ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _items = weakref . WeakKeyDictionary ( ) <EOL> def add ( self , value ) : <EOL> self . _items [ value ] = True <EOL> def __iter__ ( self ) : <EOL> return iter ( self . _items ) <EOL> class TimeoutError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class ReactorStopped ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class ResultRegistry ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , reactor ) : <EOL> self . _results = WeakSet ( ) <EOL> self . _stopped = False <EOL> self . _lock = threading . Lock ( ) <EOL> @ synchronized <EOL> def register ( self , result ) : <EOL> """<STR_LIT>""" <EOL> if self . _stopped : <EOL> raise ReactorStopped ( ) <EOL> self . _results . add ( result ) <EOL> @ synchronized <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _stopped = True <EOL> for result in self . _results : <EOL> result . _set_result ( Failure ( ReactorStopped ( ) ) ) <EOL> class EventualResult ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , deferred , _reactor ) : <EOL> """<STR_LIT>""" <EOL> self . _deferred = deferred <EOL> self . _reactor = _reactor <EOL> self . _value = None <EOL> self . _result_retrieved = False <EOL> self . _result_set = threading . Event ( ) <EOL> if deferred is not None : <EOL> self . _connect_deferred ( deferred ) <EOL> def _connect_deferred ( self , deferred ) : <EOL> """<STR_LIT>""" <EOL> self . _deferred = deferred <EOL> def put ( result , eventual = weakref . ref ( self ) ) : <EOL> eventual = eventual ( ) <EOL> if eventual : <EOL> eventual . _set_result ( result ) <EOL> else : <EOL> err ( result , "<STR_LIT>" ) <EOL> deferred . addBoth ( put ) <EOL> def _set_result ( self , result ) : <EOL> """<STR_LIT>""" <EOL> if self . _result_set . isSet ( ) : <EOL> return <EOL> self . _value = result <EOL> self . _result_set . set ( ) <EOL> def __del__ ( self ) : <EOL> if self . _result_retrieved or not self . _result_set . isSet ( ) : <EOL> return <EOL> if isinstance ( self . _value , Failure ) : <EOL> err ( self . _value , "<STR_LIT>" ) <EOL> def cancel ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _reactor . callFromThread ( lambda : self . _deferred . cancel ( ) ) <EOL> def _result ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if timeout is None : <EOL> warnings . warn ( "<STR_LIT>" , <EOL> DeprecationWarning , stacklevel = <NUM_LIT:3> ) <EOL> timeout = <NUM_LIT:2> ** <NUM_LIT> <EOL> self . _result_set . wait ( timeout ) <EOL> if not self . _result_set . is_set ( ) : <EOL> raise TimeoutError ( ) <EOL> self . _result_retrieved = True <EOL> return self . _value <EOL> def wait ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if threadable . isInIOThread ( ) : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" ) <EOL> if imp . lock_held ( ) : <EOL> try : <EOL> imp . release_lock ( ) <EOL> except RuntimeError : <EOL> pass <EOL> else : <EOL> raise RuntimeError ( <EOL> "<STR_LIT>" ) <EOL> result = self . _result ( timeout ) <EOL> if isinstance ( result , Failure ) : <EOL> result . raiseException ( ) <EOL> return result <EOL> def stash ( self ) : <EOL> """<STR_LIT>""" <EOL> return _store . store ( self ) <EOL> def original_failure ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> result = self . _result ( <NUM_LIT:0.0> ) <EOL> except TimeoutError : <EOL> return None <EOL> if isinstance ( result , Failure ) : <EOL> return result <EOL> else : <EOL> return None <EOL> class ThreadLogObserver ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , observer ) : <EOL> self . _observer = observer <EOL> if getattr ( select , "<STR_LIT>" , None ) : <EOL> from twisted . internet . pollreactor import PollReactor <EOL> reactorFactory = PollReactor <EOL> else : <EOL> from twisted . internet . selectreactor import SelectReactor <EOL> reactorFactory = SelectReactor <EOL> self . _logWritingReactor = reactorFactory ( ) <EOL> self . _logWritingReactor . _registerAsIOThread = False <EOL> self . _thread = threading . Thread ( target = self . _reader , <EOL> name = "<STR_LIT>" ) <EOL> self . _thread . start ( ) <EOL> def _reader ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _logWritingReactor . run ( installSignalHandlers = False ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _logWritingReactor . callFromThread ( self . _logWritingReactor . stop ) <EOL> def __call__ ( self , msg ) : <EOL> """<STR_LIT>""" <EOL> self . _logWritingReactor . callFromThread ( self . _observer , msg ) <EOL> class EventLoop ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , reactorFactory , atexit_register , <EOL> startLoggingWithObserver = None , <EOL> watchdog_thread = None , <EOL> reapAllProcesses = None ) : <EOL> """<STR_LIT>""" <EOL> self . _reactorFactory = reactorFactory <EOL> self . _atexit_register = atexit_register <EOL> self . _startLoggingWithObserver = startLoggingWithObserver <EOL> self . _started = False <EOL> self . _lock = threading . Lock ( ) <EOL> self . _watchdog_thread = watchdog_thread <EOL> self . _reapAllProcesses = reapAllProcesses <EOL> def _startReapingProcesses ( self ) : <EOL> """<STR_LIT>""" <EOL> lc = LoopingCall ( self . _reapAllProcesses ) <EOL> lc . clock = self . _reactor <EOL> lc . start ( <NUM_LIT:0.1> , False ) <EOL> def _common_setup ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _started = True <EOL> self . _reactor = self . _reactorFactory ( ) <EOL> self . _registry = ResultRegistry ( self . _reactor ) <EOL> self . _reactor . addSystemEventTrigger ( <EOL> "<STR_LIT>" , "<STR_LIT>" , self . _registry . stop ) <EOL> @ synchronized <EOL> def setup ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _started : <EOL> return <EOL> self . _common_setup ( ) <EOL> if platform . type == "<STR_LIT>" : <EOL> self . _reactor . callFromThread ( self . _startReapingProcesses ) <EOL> if self . _startLoggingWithObserver : <EOL> observer = ThreadLogObserver ( PythonLoggingObserver ( ) . emit ) <EOL> def start ( ) : <EOL> from twisted . python import log <EOL> original = log . showwarning <EOL> log . showwarning = warnings . showwarning <EOL> self . _startLoggingWithObserver ( observer , False ) <EOL> log . showwarning = original <EOL> self . _reactor . callFromThread ( start ) <EOL> self . _reactor . addSystemEventTrigger ( "<STR_LIT>" , "<STR_LIT>" , <EOL> observer . stop ) <EOL> t = threading . Thread ( <EOL> target = lambda : self . _reactor . run ( installSignalHandlers = False ) , <EOL> name = "<STR_LIT>" ) <EOL> t . start ( ) <EOL> self . _atexit_register ( self . _reactor . callFromThread , <EOL> self . _reactor . stop ) <EOL> self . _atexit_register ( _store . log_errors ) <EOL> if self . _watchdog_thread is not None : <EOL> self . _watchdog_thread . start ( ) <EOL> @ synchronized <EOL> def no_setup ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _started : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . _common_setup ( ) <EOL> def run_in_reactor ( self , function ) : <EOL> """<STR_LIT>""" <EOL> def runs_in_reactor ( result , args , kwargs ) : <EOL> d = maybeDeferred ( function , * args , ** kwargs ) <EOL> result . _connect_deferred ( d ) <EOL> @ wraps ( function ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> result = EventualResult ( None , self . _reactor ) <EOL> self . _registry . register ( result ) <EOL> self . _reactor . callFromThread ( runs_in_reactor , result , args , kwargs ) <EOL> return result <EOL> wrapper . wrapped_function = function <EOL> return wrapper <EOL> def wait_for_reactor ( self , function ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( "<STR_LIT>" , <EOL> DeprecationWarning , stacklevel = <NUM_LIT:2> ) <EOL> return self . wait_for ( <NUM_LIT:2> ** <NUM_LIT> ) ( function ) <EOL> def wait_for ( self , timeout ) : <EOL> """<STR_LIT>""" <EOL> def decorator ( function ) : <EOL> @ wraps ( function ) <EOL> def wrapper ( * args , ** kwargs ) : <EOL> @ self . run_in_reactor <EOL> def run ( ) : <EOL> return function ( * args , ** kwargs ) <EOL> eventual_result = run ( ) <EOL> try : <EOL> return eventual_result . wait ( timeout ) <EOL> except TimeoutError : <EOL> eventual_result . cancel ( ) <EOL> raise <EOL> wrapper . wrapped_function = function <EOL> return wrapper <EOL> return decorator <EOL> def in_reactor ( self , function ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( "<STR_LIT>" , <EOL> DeprecationWarning , stacklevel = <NUM_LIT:2> ) <EOL> @ self . run_in_reactor <EOL> @ wraps ( function ) <EOL> def add_reactor ( * args , ** kwargs ) : <EOL> return function ( self . _reactor , * args , ** kwargs ) <EOL> return add_reactor </s>
