<s> """<STR_LIT>""" <EOL> import importlib <EOL> import logging <EOL> import math <EOL> import multiprocessing <EOL> import os <EOL> import pkg_resources <EOL> from os import path <EOL> try : <EOL> import cProfile as profile <EOL> except ImportError : <EOL> import profile <EOL> import re <EOL> import signal <EOL> import sys <EOL> import time <EOL> from tornado import gen <EOL> from tornado import ioloop <EOL> from tornado import locks <EOL> import pika <EOL> try : <EOL> import raven <EOL> except ImportError : <EOL> raven = None <EOL> from pika import spec <EOL> from rejected import __version__ <EOL> from rejected import PYTHON26 <EOL> from rejected import data <EOL> from rejected import state <EOL> from rejected import stats <EOL> LOGGER = logging . getLogger ( __name__ ) <EOL> URI_RE = re . compile ( r'<STR_LIT>' ) <EOL> def import_consumer ( value ) : <EOL> """<STR_LIT>""" <EOL> parts = value . split ( '<STR_LIT:.>' ) <EOL> import_name = '<STR_LIT:.>' . join ( parts [ <NUM_LIT:0> : - <NUM_LIT:1> ] ) <EOL> import_handle = importlib . import_module ( import_name ) <EOL> version = None <EOL> if hasattr ( import_handle , '<STR_LIT>' ) : <EOL> version = import_handle . __version__ <EOL> elif len ( parts ) > <NUM_LIT:2> : <EOL> package_handle = importlib . import_module ( parts [ <NUM_LIT:0> ] ) <EOL> if hasattr ( package_handle , '<STR_LIT>' ) : <EOL> version = package_handle . __version__ <EOL> return getattr ( import_handle , parts [ - <NUM_LIT:1> ] ) , version <EOL> class Process ( multiprocessing . Process , state . State ) : <EOL> """<STR_LIT>""" <EOL> AMQP_APP_ID = '<STR_LIT>' % __version__ <EOL> STATE_PROCESSING = <NUM_LIT> <EOL> ACKED = '<STR_LIT>' <EOL> CLOSED_ON_COMPLETE = '<STR_LIT>' <EOL> ERROR = '<STR_LIT>' <EOL> FAILURES = '<STR_LIT>' <EOL> PROCESSED = '<STR_LIT>' <EOL> REDELIVERED = '<STR_LIT>' <EOL> TIME_SPENT = '<STR_LIT>' <EOL> TIME_WAITED = '<STR_LIT>' <EOL> MESSAGE_AGE = '<STR_LIT>' <EOL> MESSAGE_DROPPED = '<STR_LIT>' <EOL> MESSAGE_REQUEUED = '<STR_LIT>' <EOL> CONSUMER_EXCEPTION = '<STR_LIT>' <EOL> MESSAGE_EXCEPTION = '<STR_LIT>' <EOL> PROCESSING_EXCEPTION = '<STR_LIT>' <EOL> UNHANDLED_EXCEPTION = '<STR_LIT>' <EOL> HB_INTERVAL = <NUM_LIT> <EOL> QOS_PREFETCH_COUNT = <NUM_LIT:1> <EOL> QOS_PREFETCH_MULTIPLIER = <NUM_LIT> <EOL> QOS_MAX = <NUM_LIT> <EOL> MAX_ERROR_COUNT = <NUM_LIT:5> <EOL> MAX_ERROR_WINDOW = <NUM_LIT> <EOL> MAX_SHUTDOWN_WAIT = <NUM_LIT:5> <EOL> def __init__ ( self , <EOL> group = None , <EOL> target = None , <EOL> name = None , <EOL> args = ( ) , <EOL> kwargs = None ) : <EOL> if kwargs is None : <EOL> kwargs = { } <EOL> super ( Process , self ) . __init__ ( group , target , name , args , kwargs ) <EOL> self . ack = True <EOL> self . active_message = None <EOL> self . channel = None <EOL> self . config = None <EOL> self . connection = None <EOL> self . connection_id = <NUM_LIT:0> <EOL> self . connection_name = None <EOL> self . connections = None <EOL> self . consumer = None <EOL> self . consumer_lock = None <EOL> self . consumer_name = None <EOL> self . delivery_time = None <EOL> self . ioloop = None <EOL> self . last_failure = <NUM_LIT:0> <EOL> self . last_stats_time = None <EOL> self . logging_config = dict ( ) <EOL> self . message_connection_id = None <EOL> self . max_error_count = self . MAX_ERROR_COUNT <EOL> self . max_frame_size = spec . FRAME_MAX_SIZE <EOL> self . queue_name = None <EOL> self . prepend_path = None <EOL> self . sentry_client = None <EOL> self . state = self . STATE_INITIALIZING <EOL> self . state_start = time . time ( ) <EOL> self . stats = None <EOL> self . stats_queue = None <EOL> self . STATES [ <NUM_LIT> ] = '<STR_LIT>' <EOL> def ack_message ( self , delivery_tag ) : <EOL> """<STR_LIT>""" <EOL> if not self . can_respond : <EOL> LOGGER . warning ( '<STR_LIT>' ) <EOL> self . stats . incr ( self . CLOSED_ON_COMPLETE ) <EOL> return <EOL> LOGGER . debug ( '<STR_LIT>' , delivery_tag ) <EOL> self . channel . basic_ack ( delivery_tag = delivery_tag ) <EOL> self . stats . incr ( self . ACKED ) <EOL> def add_on_channel_close_callback ( self ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . channel . add_on_close_callback ( self . on_channel_closed ) <EOL> def calc_velocity ( self , values ) : <EOL> """<STR_LIT>""" <EOL> processed = ( values [ '<STR_LIT>' ] . get ( self . PROCESSED , <NUM_LIT:0> ) - <EOL> values [ '<STR_LIT>' ] . get ( self . PROCESSED , <NUM_LIT:0> ) ) <EOL> duration = time . time ( ) - self . last_stats_time <EOL> if not processed or not duration : <EOL> return <NUM_LIT:0> <EOL> velocity = float ( processed ) / float ( duration ) <EOL> LOGGER . debug ( '<STR_LIT>' , velocity ) <EOL> return velocity <EOL> @ property <EOL> def can_respond ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . channel : <EOL> return False <EOL> return self . message_connection_id == self . connection_id <EOL> def cancel_consumer_with_rabbitmq ( self ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> if self . channel and self . channel . is_open : <EOL> self . channel . basic_cancel ( consumer_tag = self . name ) <EOL> def close_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> self . connection . close ( ) <EOL> def connect_to_rabbitmq ( self , config , name ) : <EOL> """<STR_LIT>""" <EOL> self . set_state ( self . STATE_CONNECTING ) <EOL> self . connection_id += <NUM_LIT:1> <EOL> params = self . get_connection_parameters ( config [ name ] ) <EOL> LOGGER . debug ( '<STR_LIT>' , <EOL> params . host , params . port , params . virtual_host , <EOL> params . credentials . username ) <EOL> return pika . TornadoConnection ( params , <EOL> self . on_connect_open , <EOL> self . on_connect_failed , <EOL> self . on_closed , <EOL> False , <EOL> self . ioloop ) <EOL> @ staticmethod <EOL> def get_config ( cfg , number , name , connection ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> '<STR_LIT>' : cfg [ '<STR_LIT>' ] [ connection ] , <EOL> '<STR_LIT>' : connection , <EOL> '<STR_LIT>' : name , <EOL> '<STR_LIT>' : '<STR_LIT>' % ( name , os . getpid ( ) , number ) <EOL> } <EOL> def get_connection_parameters ( self , config ) : <EOL> """<STR_LIT>""" <EOL> heartbeat_interval = config . get ( '<STR_LIT>' , self . HB_INTERVAL ) <EOL> password = config . get ( '<STR_LIT:password>' , config . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> credentials = pika . PlainCredentials ( config . get ( '<STR_LIT:user>' , '<STR_LIT>' ) , <EOL> password ) <EOL> return pika . ConnectionParameters ( config . get ( '<STR_LIT:host>' , '<STR_LIT:localhost>' ) , <EOL> config . get ( '<STR_LIT:port>' , <NUM_LIT> ) , <EOL> config . get ( '<STR_LIT>' , '<STR_LIT:/>' ) , <EOL> credentials , <EOL> frame_max = self . max_frame_size , <EOL> socket_timeout = <NUM_LIT:10> , <EOL> heartbeat_interval = heartbeat_interval ) <EOL> def get_consumer ( self , cfg ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> consumer_ , version = import_consumer ( cfg [ '<STR_LIT>' ] ) <EOL> except ImportError as error : <EOL> if PYTHON26 : <EOL> LOGGER . exception ( '<STR_LIT>' , <EOL> cfg [ '<STR_LIT>' ] , error ) <EOL> else : <EOL> LOGGER . exception ( '<STR_LIT>' , <EOL> cfg [ '<STR_LIT>' ] , error , <EOL> exc_info = sys . exc_info ( ) ) <EOL> return <EOL> if version : <EOL> LOGGER . info ( '<STR_LIT>' , cfg [ '<STR_LIT>' ] , version ) <EOL> else : <EOL> LOGGER . info ( '<STR_LIT>' , cfg [ '<STR_LIT>' ] ) <EOL> settings = cfg . get ( '<STR_LIT>' , dict ( ) ) <EOL> settings [ '<STR_LIT>' ] = '<STR_LIT:.>' . join ( cfg [ '<STR_LIT>' ] . split ( '<STR_LIT:.>' ) [ <NUM_LIT:0> : - <NUM_LIT:1> ] ) <EOL> try : <EOL> return consumer_ ( settings = settings , process = self ) <EOL> except Exception as error : <EOL> if PYTHON26 : <EOL> LOGGER . exception ( '<STR_LIT>' , <EOL> cfg [ '<STR_LIT>' ] , error ) <EOL> else : <EOL> LOGGER . exception ( '<STR_LIT>' , <EOL> cfg [ '<STR_LIT>' ] , error , <EOL> exc_info = sys . exc_info ( ) ) <EOL> def get_module_data ( self ) : <EOL> modules = { } <EOL> for module_name in sys . modules . keys ( ) : <EOL> module = sys . modules [ module_name ] <EOL> if hasattr ( module , '<STR_LIT>' ) : <EOL> modules [ module_name ] = module . __version__ <EOL> elif hasattr ( module , '<STR_LIT:version>' ) : <EOL> modules [ module_name ] = module . version <EOL> else : <EOL> try : <EOL> version = self . get_version ( module_name ) <EOL> if version : <EOL> modules [ module_name ] = version <EOL> except Exception : <EOL> pass <EOL> return modules <EOL> @ staticmethod <EOL> def get_version ( module_name ) : <EOL> try : <EOL> return pkg_resources . get_distribution ( module_name ) . version <EOL> except pkg_resources . DistributionNotFound : <EOL> return None <EOL> @ gen . engine <EOL> def invoke_consumer ( self , message ) : <EOL> """<STR_LIT>""" <EOL> with ( yield self . consumer_lock . acquire ( ) ) : <EOL> if self . is_idle : <EOL> self . set_state ( self . STATE_PROCESSING ) <EOL> self . delivery_time = start_time = time . time ( ) <EOL> self . active_message = message <EOL> if message . properties . timestamp : <EOL> self . stats . add_timing ( self . MESSAGE_AGE , <EOL> start_time - <EOL> message . properties . timestamp ) <EOL> self . start_message_processing ( ) <EOL> try : <EOL> result = yield self . consumer . _execute ( message ) <EOL> except Exception as error : <EOL> if PYTHON26 : <EOL> LOGGER . exception ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> error ) <EOL> else : <EOL> LOGGER . exception ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> error , exc_info = sys . exc_info ( ) ) <EOL> result = data . MESSAGE_REQUEUE <EOL> LOGGER . debug ( '<STR_LIT>' , result ) <EOL> self . on_processed ( message , result , start_time ) <EOL> elif self . is_waiting_to_shutdown or self . is_shutting_down : <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> self . reject ( message . delivery_tag , True ) <EOL> self . on_ready_to_stop ( ) <EOL> else : <EOL> LOGGER . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> self . state_description ) <EOL> @ property <EOL> def is_processing ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . state in [ self . STATE_PROCESSING , self . STATE_STOP_REQUESTED ] <EOL> def on_channel_closed ( self , _channel , reply_code , reply_text ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . critical ( '<STR_LIT>' , reply_code , reply_text ) <EOL> del self . channel <EOL> self . on_ready_to_stop ( ) <EOL> def on_channel_open ( self , channel ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . channel = channel <EOL> self . add_on_channel_close_callback ( ) <EOL> self . setup_channel ( ) <EOL> def on_closed ( self , _unused , code , text ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . critical ( '<STR_LIT>' , <EOL> self . state_description , code , text ) <EOL> self . channel = None <EOL> if not self . is_shutting_down and not self . is_waiting_to_shutdown : <EOL> self . on_ready_to_stop ( ) <EOL> def on_connect_failed ( self , * args , ** kwargs ) : <EOL> LOGGER . critical ( '<STR_LIT>' , ( args , kwargs ) ) <EOL> self . on_ready_to_stop ( ) <EOL> def on_connect_open ( self , connection ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . connection = connection <EOL> self . open_channel ( ) <EOL> def on_message ( self , channel = None , method = None , properties = None , body = None ) : <EOL> """<STR_LIT>""" <EOL> if method . redelivered : <EOL> self . stats . incr ( self . REDELIVERED ) <EOL> try : <EOL> self . invoke_consumer ( data . Message ( channel , method , properties , <EOL> body ) ) <EOL> except AttributeError : <EOL> pass <EOL> def on_processed ( self , message , result , start_time ) : <EOL> self . stats . add_timing ( self . TIME_SPENT , time . time ( ) - start_time ) <EOL> if result == data . MESSAGE_DROP : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . reject ( message . delivery_tag , False ) <EOL> self . stats . incr ( self . MESSAGE_DROPPED ) <EOL> return <EOL> elif result == data . MESSAGE_EXCEPTION : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . reject ( message . delivery_tag , False ) <EOL> self . stats . incr ( self . MESSAGE_EXCEPTION ) <EOL> return <EOL> elif result == data . PROCESSING_EXCEPTION : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . reject ( message . delivery_tag , False ) <EOL> self . stats . incr ( self . PROCESSING_EXCEPTION ) <EOL> return <EOL> elif result == data . CONSUMER_EXCEPTION : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . reject ( message . delivery_tag , True ) <EOL> self . on_processing_error ( ) <EOL> self . stats . incr ( self . CONSUMER_EXCEPTION ) <EOL> return <EOL> elif result == data . UNHANDLED_EXCEPTION : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . reject ( message . delivery_tag , True ) <EOL> self . on_processing_error ( ) <EOL> self . stats . incr ( self . UNHANDLED_EXCEPTION ) <EOL> return <EOL> elif result == data . MESSAGE_REQUEUE : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . reject ( message . delivery_tag , True ) <EOL> self . stats . incr ( self . MESSAGE_REQUEUED ) <EOL> return <EOL> elif result == data . MESSAGE_ACK and self . ack : <EOL> self . ack_message ( message . delivery_tag ) <EOL> self . stats . incr ( self . PROCESSED ) <EOL> if self . is_waiting_to_shutdown : <EOL> return self . on_ready_to_stop ( ) <EOL> self . reset_state ( ) <EOL> def on_processing_error ( self ) : <EOL> """<STR_LIT>""" <EOL> duration = time . time ( ) - self . last_failure <EOL> if duration > self . MAX_ERROR_WINDOW : <EOL> LOGGER . info ( '<STR_LIT>' , <EOL> duration ) <EOL> self . reset_error_counter ( ) <EOL> self . stats . incr ( self . ERROR ) <EOL> self . last_failure = time . time ( ) <EOL> if self . too_many_errors : <EOL> LOGGER . critical ( '<STR_LIT>' , <EOL> self . stats [ self . ERROR ] ) <EOL> self . cancel_consumer_with_rabbitmq ( ) <EOL> self . close_connection ( ) <EOL> self . on_ready_to_stop ( ) <EOL> @ staticmethod <EOL> def on_qos_set ( frame ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( "<STR_LIT>" , frame ) <EOL> def on_ready_to_stop ( self ) : <EOL> self . set_state ( self . STATE_SHUTTING_DOWN ) <EOL> signal . signal ( signal . SIGABRT , signal . SIG_IGN ) <EOL> signal . signal ( signal . SIGINT , signal . SIG_IGN ) <EOL> signal . signal ( signal . SIGPROF , signal . SIG_IGN ) <EOL> signal . signal ( signal . SIGTERM , signal . SIG_IGN ) <EOL> if self . connection and self . connection . is_open : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . connection . close ( ) <EOL> self . stop_consumer ( ) <EOL> self . consumer_lock = None <EOL> if self . ioloop : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . ioloop . stop ( ) <EOL> self . set_state ( self . STATE_STOPPED ) <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> def on_sigprof ( self , _unused_signum , _unused_frame ) : <EOL> """<STR_LIT>""" <EOL> values = self . stats . report ( ) <EOL> self . stats_queue . put ( values , True ) <EOL> self . last_stats_time = time . time ( ) <EOL> signal . siginterrupt ( signal . SIGPROF , False ) <EOL> def open_channel ( self ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' , self . connection ) <EOL> self . connection . channel ( self . on_channel_open ) <EOL> @ property <EOL> def profile_file ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _kwargs [ '<STR_LIT>' ] : <EOL> return None <EOL> if os . path . exists ( self . _kwargs [ '<STR_LIT>' ] ) and os . path . isdir ( self . _kwargs [ '<STR_LIT>' ] ) : <EOL> return '<STR_LIT>' % ( path . normpath ( self . _kwargs [ '<STR_LIT>' ] ) , <EOL> os . getpid ( ) , <EOL> self . _kwargs [ '<STR_LIT>' ] ) <EOL> return None <EOL> @ property <EOL> def qos_prefetch ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . config . get ( '<STR_LIT>' , self . QOS_PREFETCH_COUNT ) <EOL> def reject ( self , delivery_tag , requeue = True ) : <EOL> """<STR_LIT>""" <EOL> if not self . ack : <EOL> raise RuntimeError ( '<STR_LIT>' ) <EOL> if not self . can_respond : <EOL> LOGGER . warning ( '<STR_LIT>' ) <EOL> self . stats . incr ( self . CLOSED_ON_COMPLETE ) <EOL> if self . is_processing : <EOL> self . reset_state ( ) <EOL> return <EOL> LOGGER . warning ( '<STR_LIT>' , delivery_tag , '<STR_LIT>' <EOL> if requeue else '<STR_LIT>' ) <EOL> self . channel . basic_nack ( delivery_tag = delivery_tag , requeue = requeue ) <EOL> if self . is_processing : <EOL> self . reset_state ( ) <EOL> def reset_error_counter ( self ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . stats [ self . ERROR ] = <NUM_LIT:0> <EOL> def reset_state ( self ) : <EOL> """<STR_LIT>""" <EOL> self . active_message = None <EOL> if self . is_waiting_to_shutdown : <EOL> self . set_state ( self . STATE_SHUTTING_DOWN ) <EOL> self . on_ready_to_stop ( ) <EOL> elif self . is_processing : <EOL> self . set_state ( self . STATE_IDLE ) <EOL> elif self . is_idle or self . is_connecting : <EOL> pass <EOL> else : <EOL> LOGGER . critical ( '<STR_LIT>' , self . state_description ) <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . profile_file : <EOL> LOGGER . info ( '<STR_LIT>' , self . profile_file ) <EOL> profile . runctx ( '<STR_LIT>' , globals ( ) , locals ( ) , <EOL> self . profile_file ) <EOL> else : <EOL> self . _run ( ) <EOL> LOGGER . debug ( '<STR_LIT>' , self . name , os . getpid ( ) , <EOL> os . getppid ( ) ) <EOL> def _run ( self ) : <EOL> """<STR_LIT>""" <EOL> self . ioloop = ioloop . IOLoop . current ( ) <EOL> self . consumer_lock = locks . Lock ( ) <EOL> try : <EOL> self . setup ( self . _kwargs [ '<STR_LIT>' ] , <EOL> self . _kwargs [ '<STR_LIT>' ] , <EOL> self . _kwargs [ '<STR_LIT>' ] , <EOL> self . _kwargs [ '<STR_LIT>' ] ) <EOL> except ( AttributeError , ImportError ) as error : <EOL> name = self . _kwargs [ '<STR_LIT>' ] <EOL> class_name = self . _kwargs [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ name ] [ '<STR_LIT>' ] <EOL> if PYTHON26 : <EOL> LOGGER . exception ( '<STR_LIT>' , <EOL> class_name , error ) <EOL> else : <EOL> LOGGER . exception ( '<STR_LIT>' , <EOL> class_name , error , exc_info = sys . exc_info ( ) ) <EOL> os . kill ( os . getppid ( ) , signal . SIGABRT ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> self . ioloop . add_callback ( self . connect_to_rabbitmq , <EOL> self . connections , <EOL> self . connection_name ) <EOL> if not self . is_stopped : <EOL> try : <EOL> self . ioloop . start ( ) <EOL> except KeyboardInterrupt : <EOL> LOGGER . warning ( '<STR_LIT>' ) <EOL> def send_exception_to_sentry ( self , exc_info ) : <EOL> """<STR_LIT>""" <EOL> if not self . sentry_client : <EOL> return <EOL> message = dict ( self . active_message ) <EOL> try : <EOL> duration = math . ceil ( time . time ( ) - self . delivery_time ) * <NUM_LIT:1000> <EOL> except TypeError : <EOL> duration = <NUM_LIT:0> <EOL> kwargs = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . get_module_data ( ) , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : self . consumer_name , <EOL> '<STR_LIT>' : self . connection_name , <EOL> '<STR_LIT>' : self . strip_uri_passwords ( dict ( os . environ ) ) , <EOL> '<STR_LIT:message>' : message } , <EOL> '<STR_LIT>' : duration } <EOL> LOGGER . debug ( '<STR_LIT>' , kwargs ) <EOL> self . sentry_client . captureException ( exc_info , ** kwargs ) <EOL> def setup ( self , cfg , consumer_name , connection_name , stats_queue ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . info ( '<STR_LIT>' , self . name , <EOL> connection_name ) <EOL> if raven and '<STR_LIT>' in cfg : <EOL> options = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> cfg [ '<STR_LIT>' ] [ consumer_name ] [ '<STR_LIT>' ] ] , <EOL> } <EOL> self . sentry_client = raven . Client ( cfg [ '<STR_LIT>' ] , ** options ) <EOL> self . connection_name = connection_name <EOL> self . consumer_name = consumer_name <EOL> self . config = cfg [ '<STR_LIT>' ] [ consumer_name ] <EOL> self . connections = cfg [ '<STR_LIT>' ] <EOL> self . consumer = self . get_consumer ( self . config ) <EOL> self . stats_queue = stats_queue <EOL> if not self . consumer : <EOL> LOGGER . critical ( '<STR_LIT>' ) <EOL> self . set_state ( self . STATE_STOPPED ) <EOL> self . on_ready_to_stop ( ) <EOL> return <EOL> self . stats = stats . Stats ( self . name , consumer_name , cfg [ '<STR_LIT>' ] or { } ) <EOL> if self . stats . statsd : <EOL> try : <EOL> self . consumer . _set_statsd ( self . stats . statsd ) <EOL> except AttributeError : <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> self . ack = self . config . get ( '<STR_LIT>' , True ) <EOL> self . max_error_count = int ( self . config . get ( '<STR_LIT>' , <EOL> self . MAX_ERROR_COUNT ) ) <EOL> self . max_frame_size = self . config . get ( '<STR_LIT>' , <EOL> spec . FRAME_MAX_SIZE ) <EOL> self . queue_name = self . config [ '<STR_LIT>' ] <EOL> self . reset_error_counter ( ) <EOL> self . setup_sighandlers ( ) <EOL> def setup_channel ( self ) : <EOL> """<STR_LIT>""" <EOL> self . set_state ( self . STATE_IDLE ) <EOL> try : <EOL> self . consumer . _set_channel ( self . channel ) <EOL> except AttributeError : <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> self . channel . basic_qos ( self . on_qos_set , <NUM_LIT:0> , self . qos_prefetch , False ) <EOL> self . channel . basic_consume ( consumer_callback = self . on_message , <EOL> queue = self . queue_name , <EOL> no_ack = not self . ack , <EOL> consumer_tag = self . name ) <EOL> def setup_sighandlers ( self ) : <EOL> """<STR_LIT>""" <EOL> signal . signal ( signal . SIGINT , signal . SIG_IGN ) <EOL> signal . signal ( signal . SIGTERM , signal . SIG_IGN ) <EOL> signal . signal ( signal . SIGPROF , self . on_sigprof ) <EOL> signal . signal ( signal . SIGABRT , self . stop ) <EOL> signal . siginterrupt ( signal . SIGPROF , False ) <EOL> signal . siginterrupt ( signal . SIGABRT , False ) <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> def start_message_processing ( self ) : <EOL> """<STR_LIT>""" <EOL> self . message_connection_id = self . connection_id <EOL> @ staticmethod <EOL> def strip_uri_passwords ( values ) : <EOL> for key in values . keys ( ) : <EOL> matches = URI_RE . search ( values [ key ] ) <EOL> if matches : <EOL> values [ key ] = values [ key ] . replace ( matches . group ( <NUM_LIT:1> ) , '<STR_LIT>' ) <EOL> return values <EOL> def stop ( self , signum = None , _unused = None ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' , self . state_description ) <EOL> if self . is_stopped : <EOL> LOGGER . warning ( '<STR_LIT>' ) <EOL> return <EOL> elif self . is_shutting_down : <EOL> LOGGER . warning ( '<STR_LIT>' ) <EOL> return <EOL> elif self . is_waiting_to_shutdown : <EOL> LOGGER . warning ( '<STR_LIT>' ) <EOL> return <EOL> self . cancel_consumer_with_rabbitmq ( ) <EOL> if self . is_processing : <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> self . set_state ( self . STATE_STOP_REQUESTED ) <EOL> if signum == signal . SIGTERM : <EOL> signal . siginterrupt ( signal . SIGTERM , False ) <EOL> return <EOL> if self . stats . statsd : <EOL> self . stats . statsd . stop ( ) <EOL> self . on_ready_to_stop ( ) <EOL> def stop_consumer ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> self . consumer . shutdown ( ) <EOL> except AttributeError : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> @ property <EOL> def too_many_errors ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . stats [ self . ERROR ] >= self . max_error_count </s>
