<s> import collections <EOL> import threading <EOL> import kombu <EOL> from kombu import exceptions as kombu_exceptions <EOL> import six <EOL> from taskflow . engines . worker_based import dispatcher <EOL> from taskflow import logging <EOL> LOG = logging . getLogger ( __name__ ) <EOL> DRAIN_EVENTS_PERIOD = <NUM_LIT:1> <EOL> _ConnectionDetails = collections . namedtuple ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> _TransportDetails = collections . namedtuple ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class Proxy ( object ) : <EOL> """<STR_LIT>""" <EOL> DEFAULT_RETRY_OPTIONS = { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:3> , <EOL> } <EOL> """<STR_LIT>""" <EOL> _RETRY_INT_OPTS = frozenset ( [ '<STR_LIT>' ] ) <EOL> def __init__ ( self , topic , exchange , <EOL> type_handlers = None , on_wait = None , url = None , <EOL> transport = None , transport_options = None , <EOL> retry_options = None ) : <EOL> self . _topic = topic <EOL> self . _exchange_name = exchange <EOL> self . _on_wait = on_wait <EOL> self . _running = threading . Event ( ) <EOL> self . _dispatcher = dispatcher . TypeDispatcher ( <EOL> requeue_filters = [ lambda data , message : not self . is_running ] , <EOL> type_handlers = type_handlers ) <EOL> ensure_options = self . DEFAULT_RETRY_OPTIONS . copy ( ) <EOL> if retry_options is not None : <EOL> for k in set ( six . iterkeys ( ensure_options ) ) : <EOL> if k in retry_options : <EOL> val = retry_options [ k ] <EOL> if k in self . _RETRY_INT_OPTS : <EOL> tmp_val = int ( val ) <EOL> else : <EOL> tmp_val = float ( val ) <EOL> if tmp_val < <NUM_LIT:0> : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( k , val ) ) <EOL> ensure_options [ k ] = tmp_val <EOL> self . _ensure_options = ensure_options <EOL> self . _drain_events_timeout = DRAIN_EVENTS_PERIOD <EOL> if transport == '<STR_LIT>' and transport_options : <EOL> polling_interval = transport_options . get ( '<STR_LIT>' ) <EOL> if polling_interval is not None : <EOL> self . _drain_events_timeout = polling_interval <EOL> self . _conn = kombu . Connection ( url , transport = transport , <EOL> transport_options = transport_options ) <EOL> self . _exchange = kombu . Exchange ( name = self . _exchange_name , <EOL> durable = False , auto_delete = True ) <EOL> @ property <EOL> def dispatcher ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _dispatcher <EOL> @ property <EOL> def connection_details ( self ) : <EOL> """<STR_LIT>""" <EOL> driver_version = self . _conn . transport . driver_version ( ) <EOL> if driver_version and driver_version . lower ( ) == '<STR_LIT>' : <EOL> driver_version = None <EOL> if self . _conn . transport_options : <EOL> transport_options = self . _conn . transport_options . copy ( ) <EOL> else : <EOL> transport_options = { } <EOL> transport = _TransportDetails ( <EOL> options = transport_options , <EOL> driver_type = self . _conn . transport . driver_type , <EOL> driver_name = self . _conn . transport . driver_name , <EOL> driver_version = driver_version ) <EOL> return _ConnectionDetails ( <EOL> uri = self . _conn . as_uri ( include_password = False ) , <EOL> transport = transport ) <EOL> @ property <EOL> def is_running ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _running . is_set ( ) <EOL> def _make_queue ( self , routing_key , exchange , channel = None ) : <EOL> """<STR_LIT>""" <EOL> queue_name = "<STR_LIT>" % ( self . _exchange_name , routing_key ) <EOL> return kombu . Queue ( name = queue_name , <EOL> routing_key = routing_key , durable = False , <EOL> exchange = exchange , auto_delete = True , <EOL> channel = channel ) <EOL> def publish ( self , msg , routing_key , reply_to = None , correlation_id = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( routing_key , six . string_types ) : <EOL> routing_keys = [ routing_key ] <EOL> else : <EOL> routing_keys = routing_key <EOL> routing_keys = [ r_k for r_k in routing_keys if r_k ] <EOL> if not routing_keys : <EOL> LOG . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" , msg , <EOL> self . _exchange_name ) <EOL> return <EOL> def _publish ( producer , routing_key ) : <EOL> queue = self . _make_queue ( routing_key , self . _exchange ) <EOL> producer . publish ( body = msg . to_dict ( ) , <EOL> routing_key = routing_key , <EOL> exchange = self . _exchange , <EOL> declare = [ queue ] , <EOL> type = msg . TYPE , <EOL> reply_to = reply_to , <EOL> correlation_id = correlation_id ) <EOL> def _publish_errback ( exc , interval ) : <EOL> LOG . exception ( '<STR_LIT>' , exc ) <EOL> LOG . info ( '<STR_LIT>' , interval ) <EOL> LOG . debug ( "<STR_LIT>" , <EOL> msg , routing_keys ) <EOL> with kombu . connections [ self . _conn ] . acquire ( block = True ) as conn : <EOL> with conn . Producer ( ) as producer : <EOL> ensure_kwargs = self . _ensure_options . copy ( ) <EOL> ensure_kwargs [ '<STR_LIT>' ] = _publish_errback <EOL> safe_publish = conn . ensure ( producer , _publish , ** ensure_kwargs ) <EOL> for routing_key in routing_keys : <EOL> safe_publish ( producer , routing_key ) <EOL> def start ( self ) : <EOL> """<STR_LIT>""" <EOL> def _drain ( conn , timeout ) : <EOL> try : <EOL> conn . drain_events ( timeout = timeout ) <EOL> except kombu_exceptions . TimeoutError : <EOL> pass <EOL> def _drain_errback ( exc , interval ) : <EOL> LOG . exception ( '<STR_LIT>' , exc ) <EOL> LOG . info ( '<STR_LIT>' , interval ) <EOL> LOG . info ( "<STR_LIT>" , <EOL> self . _exchange_name ) <EOL> with kombu . connections [ self . _conn ] . acquire ( block = True ) as conn : <EOL> queue = self . _make_queue ( self . _topic , self . _exchange , channel = conn ) <EOL> callbacks = [ self . _dispatcher . on_message ] <EOL> with conn . Consumer ( queues = queue , callbacks = callbacks ) as consumer : <EOL> ensure_kwargs = self . _ensure_options . copy ( ) <EOL> ensure_kwargs [ '<STR_LIT>' ] = _drain_errback <EOL> safe_drain = conn . ensure ( consumer , _drain , ** ensure_kwargs ) <EOL> self . _running . set ( ) <EOL> try : <EOL> while self . _running . is_set ( ) : <EOL> safe_drain ( conn , self . _drain_events_timeout ) <EOL> if self . _on_wait is not None : <EOL> self . _on_wait ( ) <EOL> finally : <EOL> self . _running . clear ( ) <EOL> def wait ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _running . wait ( ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _running . clear ( ) </s>
