<s> from __future__ import absolute_import <EOL> import sys <EOL> from contextlib import contextmanager <EOL> from collections import namedtuple <EOL> import logging <EOL> import kombu <EOL> import kombu . mixins <EOL> import kombu . pools <EOL> from lymph . events . base import BaseEventSystem <EOL> from lymph . core . events import Event <EOL> from lymph . utils . logging import setup_logger <EOL> logger = logging . getLogger ( __name__ ) <EOL> DEFAULT_SERIALIZER = '<STR_LIT>' <EOL> DEFAULT_EXCHANGE = '<STR_LIT>' <EOL> DEFAULT_MAX_RETRIES = <NUM_LIT:3> <EOL> RETRY_HEADER = '<STR_LIT>' <EOL> QueueInfo = namedtuple ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> class EventConsumer ( kombu . mixins . ConsumerMixin ) : <EOL> def __init__ ( self , event_system , connection , queue , handler , exchange , max_retries = DEFAULT_MAX_RETRIES ) : <EOL> self . connection = connection <EOL> self . queue = queue <EOL> self . handler = handler <EOL> self . exchange = exchange <EOL> self . greenlet = None <EOL> self . event_system = event_system <EOL> self . connect_max_retries = max_retries <EOL> self . retry_producer = EventProducer ( <EOL> exchange = self . event_system . retry_exchange , <EOL> routing_key = handler . queue_name , <EOL> event_system = event_system , <EOL> ) <EOL> def get_consumers ( self , Consumer , channel ) : <EOL> return [ Consumer ( queues = [ self . queue ] , callbacks = [ self . on_kombu_message ] ) ] <EOL> def create_connection ( self ) : <EOL> return kombu . pools . connections [ self . connection ] . acquire ( ) <EOL> def on_connection_revived ( self ) : <EOL> self . _declare ( ) <EOL> def _declare ( self ) : <EOL> with self . establish_connection ( ) as conn : <EOL> self . event_system . safe_declare ( conn , self . queue ) <EOL> for event_type in self . handler . event_types : <EOL> self . queue ( conn ) . bind_to ( exchange = self . exchange , routing_key = event_type ) <EOL> self . queue ( conn ) . bind_to ( exchange = self . event_system . retry_exchange , routing_key = self . handler . queue_name ) <EOL> def on_kombu_message ( self , body , message ) : <EOL> logger . debug ( "<STR_LIT>" , body ) <EOL> if self . handler . sequential : <EOL> self . _handle_message ( body , message ) <EOL> else : <EOL> self . event_system . container . spawn ( self . _handle_message , body , message ) <EOL> def _handle_message ( self , body , message ) : <EOL> try : <EOL> event = Event . deserialize ( body ) <EOL> self . handler ( event ) <EOL> message . ack ( ) <EOL> except : <EOL> self . _handle_fail ( message , event ) <EOL> finally : <EOL> if self . handler . once : <EOL> self . event_system . unsubscribe ( self . handler ) <EOL> def _handle_fail ( self , message , event ) : <EOL> retry = event . headers . get ( RETRY_HEADER , self . handler . retry ) - <NUM_LIT:1> <EOL> if retry >= <NUM_LIT:0> : <EOL> self . _requeue ( event , retry ) <EOL> message . reject ( ) <EOL> logger . exception ( '<STR_LIT>' , self . handler . queue_name , retry ) <EOL> self . event_system . container . error_hook ( sys . exc_info ( ) ) <EOL> def _requeue ( self , event , retry ) : <EOL> event . headers [ RETRY_HEADER ] = retry <EOL> self . retry_producer . emit ( event ) <EOL> def start ( self ) : <EOL> if self . greenlet : <EOL> return <EOL> self . should_stop = False <EOL> self . _declare ( ) <EOL> self . greenlet = self . event_system . container . spawn ( self . run ) <EOL> def stop ( self , ** kwargs ) : <EOL> if not self . greenlet : <EOL> return <EOL> self . should_stop = True <EOL> self . greenlet . join ( ) <EOL> self . greenlet = None <EOL> class EventProducer ( object ) : <EOL> def __init__ ( self , exchange , routing_key , event_system , max_retries = DEFAULT_MAX_RETRIES ) : <EOL> self . routing_key = routing_key <EOL> self . exchange = exchange <EOL> self . event_system = event_system <EOL> self . max_retries = max_retries <EOL> @ contextmanager <EOL> def _get_connection ( self ) : <EOL> with self . event_system . get_connection ( ) as conn : <EOL> yield conn <EOL> def _get_producer ( self , conn ) : <EOL> return conn . Producer ( <EOL> serializer = self . event_system . serializer , routing_key = self . routing_key , <EOL> exchange = self . exchange ) <EOL> def emit ( self , event ) : <EOL> with self . _get_connection ( ) as conn : <EOL> producer = self . _get_producer ( conn ) <EOL> return producer . publish ( event . serialize ( ) , retry_policy = { '<STR_LIT>' : self . max_retries } ) <EOL> class EventProducerWithDelay ( EventProducer ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , delay , * args , ** kwargs ) : <EOL> super ( EventProducerWithDelay , self ) . __init__ ( * args , ** kwargs ) <EOL> self . delay = delay <EOL> self . _intermediate_queue = None <EOL> def _get_producer ( self , conn ) : <EOL> if self . _intermediate_queue is None or self . _intermediate_queue . master != conn . as_uri ( ) : <EOL> queue = self . _prepare_intermediate_queue ( conn ) <EOL> self . _intermediate_queue = QueueInfo ( master = conn . as_uri ( ) , queue = queue ) <EOL> return super ( EventProducerWithDelay , self ) . _get_producer ( conn ) <EOL> def _prepare_intermediate_queue ( self , conn ) : <EOL> queue_name = '<STR_LIT>' % ( self . routing_key , self . delay ) <EOL> queue = self . event_system . get_queue ( queue_name , durable = False , queue_arguments = { <EOL> '<STR_LIT>' : self . event_system . exchange . name , <EOL> '<STR_LIT>' : self . routing_key , <EOL> '<STR_LIT>' : self . delay , <EOL> } ) <EOL> self . event_system . safe_declare ( conn , queue ) <EOL> queue ( conn ) . bind_to ( exchange = self . exchange , routing_key = self . routing_key ) <EOL> return queue <EOL> class KombuEventSystem ( BaseEventSystem ) : <EOL> def __init__ ( self , connection , exchange_name , serializer = DEFAULT_SERIALIZER , connect_max_retries = DEFAULT_MAX_RETRIES ) : <EOL> super ( KombuEventSystem , self ) . __init__ ( ) <EOL> self . connection = connection <EOL> self . exchange_name = exchange_name <EOL> self . exchange = kombu . Exchange ( exchange_name , '<STR_LIT>' , durable = True ) <EOL> self . waiting_exchange = kombu . Exchange ( '<STR_LIT>' % exchange_name , '<STR_LIT>' , durable = True ) <EOL> self . retry_exchange = kombu . Exchange ( '<STR_LIT>' % exchange_name , '<STR_LIT>' , durable = True ) <EOL> self . serializer = serializer <EOL> self . connect_max_retries = connect_max_retries <EOL> self . _producers = { } <EOL> self . consumers_by_queue = { } <EOL> @ classmethod <EOL> def from_config ( cls , config , ** kwargs ) : <EOL> exchange_name = config . get ( '<STR_LIT>' , DEFAULT_EXCHANGE ) <EOL> serializer = config . get ( '<STR_LIT>' , DEFAULT_SERIALIZER ) <EOL> connection = kombu . Connection ( ** config ) <EOL> return cls ( connection , exchange_name , serializer = serializer , ** kwargs ) <EOL> def on_start ( self ) : <EOL> setup_logger ( '<STR_LIT>' ) <EOL> with self . get_connection ( ) as conn : <EOL> self . exchange ( conn ) . declare ( ) <EOL> self . waiting_exchange ( conn ) . declare ( ) <EOL> self . retry_exchange ( conn ) . declare ( ) <EOL> def on_stop ( self , ** kwargs ) : <EOL> for consumer in self . consumers_by_queue . values ( ) : <EOL> consumer . stop ( ** kwargs ) <EOL> self . consumers_by_queue . clear ( ) <EOL> def subscribe ( self , handler , consume = True ) : <EOL> try : <EOL> consumer = self . consumers_by_queue [ handler . queue_name ] <EOL> except KeyError : <EOL> consumer = self . setup_consumer ( handler ) <EOL> else : <EOL> if consumer . handler != handler : <EOL> raise RuntimeError ( '<STR_LIT>' % handler . queue_name ) <EOL> if consume : <EOL> consumer . start ( ) <EOL> return consumer <EOL> def unsubscribe ( self , handler ) : <EOL> queue_name = handler . queue_name <EOL> try : <EOL> consumer = self . consumers_by_queue [ queue_name ] <EOL> except KeyError : <EOL> raise KeyError ( '<STR_LIT>' % queue_name ) <EOL> if consumer . handler != handler : <EOL> raise KeyError ( '<STR_LIT>' % ( handler , queue_name ) ) <EOL> consumer . stop ( ) <EOL> del self . consumers_by_queue [ queue_name ] <EOL> def setup_consumer ( self , handler ) : <EOL> if handler . broadcast : <EOL> queue = self . get_queue ( handler . queue_name , auto_delete = True , durable = False ) <EOL> else : <EOL> queue = self . get_queue ( handler . queue_name , auto_delete = handler . once , durable = False ) <EOL> consumer = EventConsumer ( self , self . connection , queue , handler , self . exchange , max_retries = self . connect_max_retries ) <EOL> self . consumers_by_queue [ handler . queue_name ] = consumer <EOL> return consumer <EOL> @ contextmanager <EOL> def get_connection ( self ) : <EOL> with kombu . pools . connections [ self . connection ] . acquire ( ) as conn : <EOL> conn . ensure_connection ( max_retries = self . connect_max_retries ) <EOL> logger . debug ( '<STR_LIT>' , conn . as_uri ( ) ) <EOL> yield conn <EOL> @ staticmethod <EOL> def get_queue ( name , ** kwargs ) : <EOL> queue_arguments = kwargs . pop ( '<STR_LIT>' , { } ) <EOL> queue_arguments [ '<STR_LIT>' ] = '<STR_LIT:all>' <EOL> return kombu . Queue ( name , queue_arguments = queue_arguments , ** kwargs ) <EOL> @ staticmethod <EOL> def safe_declare ( conn , queue ) : <EOL> try : <EOL> queue ( conn ) . declare ( ) <EOL> except conn . connection . channel_errors : <EOL> queue ( conn ) . declare ( ) <EOL> def emit ( self , event , delay = <NUM_LIT:0> ) : <EOL> producer = self . _get_producer ( event . evt_type , delay ) <EOL> producer . emit ( event ) <EOL> def _get_producer ( self , event_type , delay = <NUM_LIT:0> ) : <EOL> try : <EOL> return self . _producers [ event_type , delay ] <EOL> except KeyError : <EOL> if delay : <EOL> producer = EventProducerWithDelay ( <EOL> delay = int ( <NUM_LIT:1000> * delay ) , <EOL> exchange = self . waiting_exchange , <EOL> routing_key = event_type , <EOL> event_system = self , <EOL> ) <EOL> else : <EOL> producer = EventProducer ( <EOL> exchange = self . exchange , <EOL> routing_key = event_type , <EOL> event_system = self , <EOL> ) <EOL> self . _producers [ event_type , delay ] = producer <EOL> return producer </s>
