<s> import logging <EOL> import random <EOL> import sys <EOL> from gearman import compat <EOL> from gearman . connection_manager import GearmanConnectionManager <EOL> from gearman . worker_handler import GearmanWorkerCommandHandler <EOL> from gearman . errors import ConnectionError <EOL> gearman_logger = logging . getLogger ( __name__ ) <EOL> POLL_TIMEOUT_IN_SECONDS = <NUM_LIT> <EOL> class GearmanWorker ( GearmanConnectionManager ) : <EOL> """<STR_LIT>""" <EOL> command_handler_class = GearmanWorkerCommandHandler <EOL> def __init__ ( self , host_list = None ) : <EOL> super ( GearmanWorker , self ) . __init__ ( host_list = host_list ) <EOL> self . randomized_connections = None <EOL> self . worker_abilities = { } <EOL> self . worker_client_id = None <EOL> self . command_handler_holding_job_lock = None <EOL> self . _update_initial_state ( ) <EOL> def _update_initial_state ( self ) : <EOL> self . handler_initial_state [ '<STR_LIT>' ] = self . worker_abilities . keys ( ) <EOL> self . handler_initial_state [ '<STR_LIT>' ] = self . worker_client_id <EOL> def register_task ( self , task , callback_function ) : <EOL> """<STR_LIT>""" <EOL> self . worker_abilities [ task ] = callback_function <EOL> self . _update_initial_state ( ) <EOL> for command_handler in self . handler_to_connection_map . iterkeys ( ) : <EOL> command_handler . set_abilities ( self . handler_initial_state [ '<STR_LIT>' ] ) <EOL> return task <EOL> def unregister_task ( self , task ) : <EOL> """<STR_LIT>""" <EOL> self . worker_abilities . pop ( task , None ) <EOL> self . _update_initial_state ( ) <EOL> for command_handler in self . handler_to_connection_map . iterkeys ( ) : <EOL> command_handler . set_abilities ( self . handler_initial_state [ '<STR_LIT>' ] ) <EOL> return task <EOL> def set_client_id ( self , client_id ) : <EOL> """<STR_LIT>""" <EOL> self . worker_client_id = client_id <EOL> self . _update_initial_state ( ) <EOL> for command_handler in self . handler_to_connection_map . iterkeys ( ) : <EOL> command_handler . set_client_id ( self . handler_initial_state [ '<STR_LIT>' ] ) <EOL> return client_id <EOL> def work ( self , poll_timeout = POLL_TIMEOUT_IN_SECONDS ) : <EOL> """<STR_LIT>""" <EOL> continue_working = True <EOL> worker_connections = [ ] <EOL> had_job = [ ] <EOL> def continue_while_connections_alive ( any_activity ) : <EOL> if had_job and not self . has_job_lock ( ) : <EOL> return self . after_poll ( any_activity ) and self . after_job ( ) <EOL> del had_job [ : ] <EOL> if self . has_job_lock ( ) : <EOL> had_job . append ( True ) <EOL> return self . after_poll ( any_activity ) <EOL> while continue_working : <EOL> worker_connections = self . establish_worker_connections ( ) <EOL> continue_working = self . poll_connections_until_stopped ( worker_connections , continue_while_connections_alive , timeout = poll_timeout ) <EOL> for current_connection in worker_connections : <EOL> current_connection . close ( ) <EOL> def shutdown ( self ) : <EOL> self . command_handler_holding_job_lock = None <EOL> super ( GearmanWorker , self ) . shutdown ( ) <EOL> def establish_worker_connections ( self ) : <EOL> """<STR_LIT>""" <EOL> self . randomized_connections = list ( self . connection_list ) <EOL> random . shuffle ( self . randomized_connections ) <EOL> output_connections = [ ] <EOL> for current_connection in self . randomized_connections : <EOL> try : <EOL> valid_connection = self . establish_connection ( current_connection ) <EOL> output_connections . append ( valid_connection ) <EOL> except ConnectionError : <EOL> pass <EOL> return output_connections <EOL> def after_poll ( self , any_activity ) : <EOL> """<STR_LIT>""" <EOL> return True <EOL> def after_job ( self ) : <EOL> """<STR_LIT>""" <EOL> return True <EOL> def handle_error ( self , current_connection ) : <EOL> """<STR_LIT>""" <EOL> current_handler = self . connection_to_handler_map . get ( current_connection ) <EOL> if current_handler : <EOL> self . set_job_lock ( current_handler , lock = False ) <EOL> super ( GearmanWorker , self ) . handle_error ( current_connection ) <EOL> def _get_handler_for_job ( self , current_job ) : <EOL> return self . connection_to_handler_map [ current_job . connection ] <EOL> def wait_until_updates_sent ( self , multiple_gearman_jobs , poll_timeout = None ) : <EOL> connection_set = set ( [ current_job . connection for current_job in multiple_gearman_jobs ] ) <EOL> def continue_while_updates_pending ( any_activity ) : <EOL> return compat . any ( current_connection . writable ( ) for current_connection in connection_set ) <EOL> self . poll_connections_until_stopped ( connection_set , continue_while_updates_pending , timeout = poll_timeout ) <EOL> def send_job_status ( self , current_job , numerator , denominator , poll_timeout = None ) : <EOL> """<STR_LIT>""" <EOL> current_handler = self . _get_handler_for_job ( current_job ) <EOL> current_handler . send_job_status ( current_job , numerator = numerator , denominator = denominator ) <EOL> self . wait_until_updates_sent ( [ current_job ] , poll_timeout = poll_timeout ) <EOL> def send_job_complete ( self , current_job , data , poll_timeout = None ) : <EOL> current_handler = self . _get_handler_for_job ( current_job ) <EOL> current_handler . send_job_complete ( current_job , data = data ) <EOL> self . wait_until_updates_sent ( [ current_job ] , poll_timeout = poll_timeout ) <EOL> def send_job_failure ( self , current_job , poll_timeout = None ) : <EOL> """<STR_LIT>""" <EOL> current_handler = self . _get_handler_for_job ( current_job ) <EOL> current_handler . send_job_failure ( current_job ) <EOL> self . wait_until_updates_sent ( [ current_job ] , poll_timeout = poll_timeout ) <EOL> def send_job_exception ( self , current_job , data , poll_timeout = None ) : <EOL> """<STR_LIT>""" <EOL> current_handler = self . _get_handler_for_job ( current_job ) <EOL> current_handler . send_job_exception ( current_job , data = data ) <EOL> current_handler . send_job_failure ( current_job ) <EOL> self . wait_until_updates_sent ( [ current_job ] , poll_timeout = poll_timeout ) <EOL> def send_job_data ( self , current_job , data , poll_timeout = None ) : <EOL> """<STR_LIT>""" <EOL> current_handler = self . _get_handler_for_job ( current_job ) <EOL> current_handler . send_job_data ( current_job , data = data ) <EOL> self . wait_until_updates_sent ( [ current_job ] , poll_timeout = poll_timeout ) <EOL> def send_job_warning ( self , current_job , data , poll_timeout = None ) : <EOL> """<STR_LIT>""" <EOL> current_handler = self . _get_handler_for_job ( current_job ) <EOL> current_handler . send_job_warning ( current_job , data = data ) <EOL> self . wait_until_updates_sent ( [ current_job ] , poll_timeout = poll_timeout ) <EOL> def create_job ( self , command_handler , job_handle , task , unique , data ) : <EOL> """<STR_LIT>""" <EOL> current_connection = self . handler_to_connection_map [ command_handler ] <EOL> return self . job_class ( current_connection , job_handle , task , unique , data ) <EOL> def on_job_execute ( self , current_job ) : <EOL> try : <EOL> function_callback = self . worker_abilities [ current_job . task ] <EOL> job_result = function_callback ( self , current_job ) <EOL> except Exception : <EOL> return self . on_job_exception ( current_job , sys . exc_info ( ) ) <EOL> return self . on_job_complete ( current_job , job_result ) <EOL> def on_job_exception ( self , current_job , exc_info ) : <EOL> self . send_job_failure ( current_job ) <EOL> return False <EOL> def on_job_complete ( self , current_job , job_result ) : <EOL> self . send_job_complete ( current_job , job_result ) <EOL> return True <EOL> def set_job_lock ( self , command_handler , lock ) : <EOL> """<STR_LIT>""" <EOL> if command_handler not in self . handler_to_connection_map : <EOL> return False <EOL> failed_lock = bool ( lock and self . command_handler_holding_job_lock is not None ) <EOL> failed_unlock = bool ( not lock and self . command_handler_holding_job_lock != command_handler ) <EOL> if failed_lock or failed_unlock : <EOL> return False <EOL> if lock : <EOL> self . command_handler_holding_job_lock = command_handler <EOL> else : <EOL> self . command_handler_holding_job_lock = None <EOL> return True <EOL> def has_job_lock ( self ) : <EOL> return bool ( self . command_handler_holding_job_lock is not None ) <EOL> def check_job_lock ( self , command_handler ) : <EOL> """<STR_LIT>""" <EOL> return bool ( self . command_handler_holding_job_lock == command_handler ) </s>
