<s> import functools <EOL> import zmq <EOL> from concurrent . futures import ThreadPoolExecutor <EOL> from logbook import Logger <EOL> from zmq . eventloop import ioloop , zmqstream <EOL> from onitu . escalator . client import Escalator , EscalatorClosed <EOL> from onitu . utils import log_traceback , get_brocker_uri , get_events_uri <EOL> from onitu . utils import cpu_count <EOL> from . responses import ERROR <EOL> ioloop . install ( ) <EOL> class Brocker ( object ) : <EOL> def __init__ ( self , session ) : <EOL> self . logger = Logger ( "<STR_LIT>" ) <EOL> self . context = zmq . Context . instance ( ) <EOL> self . escalator = Escalator ( session ) <EOL> self . session = session <EOL> self . futures = { } <EOL> self . stream = None <EOL> self . loop = None <EOL> self . pool = None <EOL> def start ( self ) : <EOL> router = None <EOL> try : <EOL> self . loop = ioloop . IOLoop . instance ( ) <EOL> self . pool = ThreadPoolExecutor ( cpu_count ( ) * <NUM_LIT:3> ) <EOL> router = self . context . socket ( zmq . ROUTER ) <EOL> router . bind ( get_brocker_uri ( self . session ) ) <EOL> self . stream = zmqstream . ZMQStream ( router , self . loop ) <EOL> self . stream . on_recv ( self . handle ) <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> self . loop . start ( ) <EOL> except zmq . ZMQError as e : <EOL> if e . errno == zmq . ETERM : <EOL> pass <EOL> else : <EOL> raise <EOL> except EscalatorClosed : <EOL> pass <EOL> except Exception : <EOL> log_traceback ( self . logger ) <EOL> finally : <EOL> if self . stream : <EOL> self . stream . close ( ) <EOL> def handle ( self , msg ) : <EOL> identity = msg [ <NUM_LIT:0> ] <EOL> args = tuple ( msg [ <NUM_LIT:1> : ] ) <EOL> if args in self . futures : <EOL> future = self . futures [ args ] <EOL> else : <EOL> future = self . pool . submit ( self . get_response , * args ) <EOL> self . futures [ args ] = future <EOL> self . loop . add_future ( <EOL> future , functools . partial ( self . reply , identity ) <EOL> ) <EOL> def reply ( self , identity , future ) : <EOL> try : <EOL> if future . exception ( ) : <EOL> self . stream . send_multipart ( [ identity ] + [ ERROR ] ) <EOL> raise future . exception ( ) <EOL> self . stream . send_multipart ( [ identity ] + future . result ( ) ) <EOL> except zmq . ZMQError as e : <EOL> if e . errno == zmq . ETERM : <EOL> self . loop . stop ( ) <EOL> else : <EOL> log_traceback ( self . logger ) <EOL> except EscalatorClosed : <EOL> self . loop . stop ( ) <EOL> except Exception : <EOL> log_traceback ( self . logger ) <EOL> finally : <EOL> self . futures = { <EOL> k : v for k , v in self . futures . items ( ) if v != future <EOL> } <EOL> def close ( self ) : <EOL> self . escalator . close ( ) <EOL> self . context . term ( ) <EOL> def get_response ( self , cmd , fid , * args ) : <EOL> for source in self . select_best_source ( fid . decode ( ) ) : <EOL> dealer = None <EOL> try : <EOL> dealer = self . context . socket ( zmq . DEALER ) <EOL> dealer . connect ( get_events_uri ( self . session , source , '<STR_LIT>' ) ) <EOL> dealer . send_multipart ( ( cmd , fid ) + args ) <EOL> response = dealer . recv_multipart ( ) <EOL> if not response or response [ <NUM_LIT:0> ] == ERROR : <EOL> self . logger . debug ( "<STR_LIT>" , source ) <EOL> continue <EOL> return response <EOL> finally : <EOL> if dealer : <EOL> dealer . close ( ) <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> return [ ERROR ] <EOL> def select_best_source ( self , fid ) : <EOL> excluded = set ( ) <EOL> while True : <EOL> services = set ( <EOL> key . split ( '<STR_LIT::>' ) [ - <NUM_LIT:1> ] for key in <EOL> self . escalator . range ( <EOL> '<STR_LIT>' . format ( fid ) , include_value = False <EOL> ) <EOL> ) <EOL> sources = services - excluded <EOL> max_velocity = <NUM_LIT:0.> <EOL> source = None <EOL> for candidate in sources : <EOL> options = self . escalator . get ( <EOL> u'<STR_LIT>' . format ( candidate ) , default = { } <EOL> ) <EOL> velocity = options . get ( '<STR_LIT>' , <NUM_LIT:0.5> ) <EOL> if velocity > max_velocity : <EOL> max_velocity = velocity <EOL> source = candidate <EOL> if not source : <EOL> break <EOL> self . logger . debug ( "<STR_LIT>" , source ) <EOL> yield source <EOL> excluded . add ( source ) </s>
