<s> """<STR_LIT>""" <EOL> __license__ = """<STR_LIT>""" <EOL> __all__ = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> from collections import namedtuple <EOL> import gevent <EOL> import gevent . event <EOL> import gevent . lock <EOL> import gevent . queue <EOL> import gevent . socket as gsocket <EOL> import logging <EOL> import socket as pysocket <EOL> import sys as _sys <EOL> import threading <EOL> import time <EOL> from . utils . compat import Queue , Empty , Semaphore <EOL> log = logging . getLogger ( __name__ ) <EOL> class ResponseFuture ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , handler ) : <EOL> """<STR_LIT>""" <EOL> self . handler = handler <EOL> self . error = False <EOL> self . _ready = handler . Event ( ) <EOL> def set_response ( self , response ) : <EOL> """<STR_LIT>""" <EOL> self . response = response <EOL> self . _ready . set ( ) <EOL> def set_error ( self , error ) : <EOL> """<STR_LIT>""" <EOL> self . error = error <EOL> self . _ready . set ( ) <EOL> def get ( self , response_cls = None , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> self . _ready . wait ( timeout ) <EOL> if self . error : <EOL> raise self . error <EOL> if response_cls : <EOL> return response_cls ( self . response ) <EOL> else : <EOL> return self . response <EOL> class Handler ( object ) : <EOL> """<STR_LIT>""" <EOL> def spawn ( self , target , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> class ThreadingHandler ( Handler ) : <EOL> """<STR_LIT>""" <EOL> Queue = Queue <EOL> Event = threading . Event <EOL> Lock = threading . Lock <EOL> Semaphore = Semaphore <EOL> Socket = pysocket <EOL> _workers_spawned = <NUM_LIT:0> <EOL> def sleep ( self , seconds = <NUM_LIT:0> ) : <EOL> time . sleep ( seconds ) <EOL> def RLock ( * args , ** kwargs ) : <EOL> kwargs [ '<STR_LIT>' ] = False <EOL> try : <EOL> return threading . RLock ( * args [ <NUM_LIT:1> : ] , ** kwargs ) <EOL> except TypeError : <EOL> kwargs . pop ( '<STR_LIT>' ) <EOL> return threading . RLock ( * args [ <NUM_LIT:1> : ] , ** kwargs ) <EOL> def spawn ( self , target , * args , ** kwargs ) : <EOL> if '<STR_LIT:name>' in kwargs : <EOL> kwargs [ '<STR_LIT:name>' ] = "<STR_LIT>" . format ( ThreadingHandler . _workers_spawned , kwargs [ '<STR_LIT:name>' ] ) <EOL> t = threading . Thread ( target = target , * args , ** kwargs ) <EOL> t . daemon = True <EOL> t . start ( ) <EOL> ThreadingHandler . _workers_spawned += <NUM_LIT:1> <EOL> return t <EOL> class GEventHandler ( Handler ) : <EOL> """<STR_LIT>""" <EOL> Queue = gevent . queue . JoinableQueue <EOL> Event = gevent . event . Event <EOL> Lock = gevent . lock . RLock <EOL> RLock = gevent . lock . RLock <EOL> Semaphore = gevent . lock . Semaphore <EOL> Socket = gsocket <EOL> def sleep ( self , seconds = <NUM_LIT:0> ) : <EOL> gevent . sleep ( seconds ) <EOL> def spawn ( self , target , * args , ** kwargs ) : <EOL> if '<STR_LIT:name>' in kwargs : <EOL> kwargs . pop ( '<STR_LIT:name>' ) <EOL> t = gevent . spawn ( target , * args , ** kwargs ) <EOL> return t <EOL> class RequestHandler ( object ) : <EOL> """<STR_LIT>""" <EOL> Task = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> Shared = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def __init__ ( self , handler , connection ) : <EOL> """<STR_LIT>""" <EOL> self . handler = handler <EOL> self . shared = self . Shared ( connection = connection , <EOL> requests = handler . Queue ( ) , <EOL> ending = handler . Event ( ) ) <EOL> def __del__ ( self ) : <EOL> self . stop ( ) <EOL> def request ( self , request , has_response = True ) : <EOL> """<STR_LIT>""" <EOL> future = None <EOL> if has_response : <EOL> future = ResponseFuture ( self . handler ) <EOL> task = self . Task ( request , future ) <EOL> self . shared . requests . put ( task ) <EOL> return future <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> self . t = self . _start_thread ( ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> shared = self . shared <EOL> self . shared = None <EOL> log . info ( "<STR_LIT>" ) <EOL> shared . requests . join ( ) <EOL> shared . ending . set ( ) <EOL> def _start_thread ( self ) : <EOL> """<STR_LIT>""" <EOL> shared = self . shared <EOL> def worker ( ) : <EOL> try : <EOL> while not shared . ending . is_set ( ) : <EOL> try : <EOL> task = shared . requests . get ( timeout = <NUM_LIT:1> ) <EOL> except Empty : <EOL> continue <EOL> try : <EOL> shared . connection . request ( task . request ) <EOL> if task . future : <EOL> res = shared . connection . response ( ) <EOL> task . future . set_response ( res ) <EOL> except Exception as e : <EOL> if task . future : <EOL> task . future . set_error ( e ) <EOL> finally : <EOL> shared . requests . task_done ( ) <EOL> log . info ( "<STR_LIT>" ) <EOL> except : <EOL> if _sys is None : <EOL> return <EOL> raise <EOL> name = "<STR_LIT>" . format ( <EOL> self . shared . connection . host , self . shared . connection . port ) <EOL> return self . handler . spawn ( worker , name = name ) </s>
