<s> '''<STR_LIT>''' <EOL> import cPickle as pickle <EOL> from threading import Thread <EOL> from collections import deque <EOL> from tornado . web import * <EOL> from tornado . ioloop import IOLoop <EOL> from traceback import format_exc <EOL> from tornado . options import options <EOL> import zmq <EOL> import logging <EOL> import zlib <EOL> from random import choice , shuffle <EOL> class EventManager ( ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , address = None ) : <EOL> self . __handlers = { } <EOL> self . address = address <EOL> self . __zmq_context = zmq . Context ( ) <EOL> self . __remote_servers = { } <EOL> self . __thread = None <EOL> self . __queued_servers = deque ( ) <EOL> def register_server ( self , address ) : <EOL> '''<STR_LIT>''' <EOL> if address in self . __remote_servers : <EOL> raise Exception ( '<STR_LIT>' , address ) <EOL> socket = self . __zmq_context . socket ( zmq . PUSH ) <EOL> socket . connect ( address ) <EOL> self . __remote_servers [ address ] = socket <EOL> self . refresh_server_queue ( ) <EOL> def remove_server ( self , address ) : <EOL> '''<STR_LIT>''' <EOL> del self . __remote_servers [ address ] <EOL> self . refresh_server_queue ( ) <EOL> def remove_all_servers ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . __remote_servers . clear ( ) <EOL> self . refresh_server_queue ( ) <EOL> def refresh_server_queue ( self ) : <EOL> '''<STR_LIT>''' <EOL> self . __queued_servers . clear ( ) <EOL> self . __queued_servers . extend ( self . __remote_servers . itervalues ( ) ) <EOL> shuffle ( self . __queued_servers ) <EOL> def register_handler ( self , event_name , event_handler , run_on_main_loop = False , request_handler = None , persist = False ) : <EOL> '''<STR_LIT>''' <EOL> if not event_name in self . __handlers : <EOL> self . __handlers [ event_name ] = set ( ) <EOL> handler_tuple = ( event_handler , run_on_main_loop , request_handler , persist ) <EOL> self . __handlers [ event_name ] . add ( handler_tuple ) <EOL> return ( event_name , handler_tuple ) <EOL> def remove_handler ( self , handler_sig ) : <EOL> '''<STR_LIT>''' <EOL> self . __handlers [ handler_sig [ <NUM_LIT:0> ] ] . discard ( handler_sig [ <NUM_LIT:1> ] ) <EOL> def start_listening ( self ) : <EOL> '''<STR_LIT>''' <EOL> if self . __thread : <EOL> return <EOL> def receive ( ) : <EOL> context = zmq . Context ( ) <EOL> socket = context . socket ( zmq . PULL ) <EOL> socket . bind ( self . address ) <EOL> while True : <EOL> event = pickle . loads ( zlib . decompress ( socket . recv ( ) ) ) <EOL> event_name = event [ '<STR_LIT:name>' ] <EOL> event_args = event [ '<STR_LIT:args>' ] <EOL> if event_name in self . __handlers : <EOL> handlers = self . __handlers [ event_name ] <EOL> for handler in list ( handlers ) : <EOL> if not handler [ <NUM_LIT:3> ] : <EOL> handlers . remove ( handler ) <EOL> try : <EOL> if handler [ <NUM_LIT:2> ] and handler [ <NUM_LIT:2> ] . _finished : <EOL> continue <EOL> if handler [ <NUM_LIT:1> ] : <EOL> ( lambda h : IOLoop . instance ( ) . add_callback ( lambda : h [ <NUM_LIT:0> ] ( event_args ) ) ) ( handler ) <EOL> else : <EOL> handler [ <NUM_LIT:0> ] ( event_args ) <EOL> except Exception as e : <EOL> logging . error ( format_exc ( ) ) <EOL> self . __thread = Thread ( target = receive ) <EOL> self . __thread . daemon = True <EOL> self . __thread . start ( ) <EOL> def send_to_server ( self , address , event_name , event_args ) : <EOL> '''<STR_LIT>''' <EOL> event = { '<STR_LIT:name>' : event_name , '<STR_LIT:args>' : event_args } <EOL> event_data = zlib . compress ( pickle . dumps ( event ) ) <EOL> self . __remote_servers [ address ] . send ( event_data ) <EOL> def send ( self , event_name , event_args , broadcast = True ) : <EOL> '''<STR_LIT>''' <EOL> if not self . __remote_servers : <EOL> return <EOL> event = { '<STR_LIT:name>' : event_name , '<STR_LIT:args>' : event_args } <EOL> event_data = zlib . compress ( pickle . dumps ( event ) ) <EOL> if not broadcast : <EOL> self . __queued_servers [ <NUM_LIT:0> ] . send ( event_data ) <EOL> self . __queued_servers . rotate ( - <NUM_LIT:1> ) <EOL> return <EOL> for socket in self . __queued_servers : <EOL> socket . send ( event_data ) <EOL> @ classmethod <EOL> def instance ( cls ) : <EOL> '''<STR_LIT>''' <EOL> if not hasattr ( cls , '<STR_LIT>' ) : <EOL> cls . _instance = cls ( ) <EOL> return cls . _instance </s>
