<s> import abc <EOL> import eventlet <EOL> import six <EOL> from kombu . mixins import ConsumerMixin <EOL> from st2common import log as logging <EOL> from st2common . util . greenpooldispatch import BufferedDispatcher <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class QueueConsumer ( ConsumerMixin ) : <EOL> def __init__ ( self , connection , queues , handler ) : <EOL> self . connection = connection <EOL> self . _dispatcher = BufferedDispatcher ( ) <EOL> self . _queues = queues <EOL> self . _handler = handler <EOL> def shutdown ( self ) : <EOL> self . _dispatcher . shutdown ( ) <EOL> def get_consumers ( self , Consumer , channel ) : <EOL> consumer = Consumer ( queues = self . _queues , accept = [ '<STR_LIT>' ] , callbacks = [ self . process ] ) <EOL> consumer . qos ( prefetch_count = <NUM_LIT:1> ) <EOL> return [ consumer ] <EOL> def process ( self , body , message ) : <EOL> try : <EOL> if not isinstance ( body , self . _handler . message_type ) : <EOL> raise TypeError ( '<STR_LIT>' % type ( body ) ) <EOL> self . _dispatcher . dispatch ( self . _process_message , body ) <EOL> except : <EOL> LOG . exception ( '<STR_LIT>' , self . __class__ . __name__ , body ) <EOL> finally : <EOL> message . ack ( ) <EOL> def _process_message ( self , body ) : <EOL> try : <EOL> self . _handler . process ( body ) <EOL> except : <EOL> LOG . exception ( '<STR_LIT>' , self . __class__ . __name__ , body ) <EOL> class StagedQueueConsumer ( QueueConsumer ) : <EOL> """<STR_LIT>""" <EOL> def process ( self , body , message ) : <EOL> try : <EOL> if not isinstance ( body , self . _handler . message_type ) : <EOL> raise TypeError ( '<STR_LIT>' % type ( body ) ) <EOL> response = self . _handler . pre_ack_process ( body ) <EOL> self . _dispatcher . dispatch ( self . _process_message , response ) <EOL> except : <EOL> LOG . exception ( '<STR_LIT>' , self . __class__ . __name__ , body ) <EOL> finally : <EOL> message . ack ( ) <EOL> @ six . add_metaclass ( abc . ABCMeta ) <EOL> class MessageHandler ( object ) : <EOL> message_type = None <EOL> def __init__ ( self , connection , queues ) : <EOL> self . _queue_consumer = self . _get_queue_consumer ( connection , queues ) <EOL> self . _consumer_thread = None <EOL> def start ( self , wait = False ) : <EOL> LOG . info ( '<STR_LIT>' , self . __class__ . __name__ ) <EOL> self . _consumer_thread = eventlet . spawn ( self . _queue_consumer . run ) <EOL> if wait : <EOL> self . wait ( ) <EOL> def wait ( self ) : <EOL> self . _consumer_thread . wait ( ) <EOL> def shutdown ( self ) : <EOL> LOG . info ( '<STR_LIT>' , self . __class__ . __name__ ) <EOL> self . _queue_consumer . shutdown ( ) <EOL> @ abc . abstractmethod <EOL> def process ( self , message ) : <EOL> pass <EOL> def _get_queue_consumer ( self , connection , queues ) : <EOL> return QueueConsumer ( connection , queues , self ) <EOL> @ six . add_metaclass ( abc . ABCMeta ) <EOL> class StagedMessageHandler ( MessageHandler ) : <EOL> """<STR_LIT>""" <EOL> @ abc . abstractmethod <EOL> def pre_ack_process ( self , message ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def _get_queue_consumer ( self , connection , queues ) : <EOL> return StagedQueueConsumer ( connection , queues , self ) </s>
