<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , unicode_literals <EOL> import numbers <EOL> import socket <EOL> from bisect import bisect <EOL> from collections import namedtuple <EOL> from contextlib import contextmanager <EOL> from time import time <EOL> from vine import promise <EOL> from kombu . exceptions import InconsistencyError , VersionMismatch <EOL> from kombu . five import Empty , values , string_t <EOL> from kombu . log import get_logger <EOL> from kombu . utils import cached_property , register_after_fork , uuid <EOL> from kombu . utils . eventio import poll , READ , ERR <EOL> from kombu . utils . encoding import bytes_to_str <EOL> from kombu . utils . json import loads , dumps <EOL> from kombu . utils . scheduling import cycle_by_name <EOL> from kombu . utils . url import _parse_url <EOL> from . import virtual <EOL> try : <EOL> import redis <EOL> except ImportError : <EOL> redis = None <EOL> try : <EOL> from redis import sentinel <EOL> except ImportError : <EOL> sentinel = None <EOL> logger = get_logger ( '<STR_LIT>' ) <EOL> crit , warn = logger . critical , logger . warn <EOL> DEFAULT_PORT = <NUM_LIT> <EOL> DEFAULT_DB = <NUM_LIT:0> <EOL> PRIORITY_STEPS = [ <NUM_LIT:0> , <NUM_LIT:3> , <NUM_LIT:6> , <NUM_LIT:9> ] <EOL> error_classes_t = namedtuple ( '<STR_LIT>' , ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> ) ) <EOL> NO_ROUTE_ERROR = """<STR_LIT>""" <EOL> def get_redis_error_classes ( ) : <EOL> from redis import exceptions <EOL> if hasattr ( exceptions , '<STR_LIT>' ) : <EOL> DataError = exceptions . InvalidData <EOL> else : <EOL> DataError = exceptions . DataError <EOL> return error_classes_t ( <EOL> ( virtual . Transport . connection_errors + ( <EOL> InconsistencyError , <EOL> socket . error , <EOL> IOError , <EOL> OSError , <EOL> exceptions . ConnectionError , <EOL> exceptions . AuthenticationError , <EOL> exceptions . TimeoutError ) ) , <EOL> ( virtual . Transport . channel_errors + ( <EOL> DataError , <EOL> exceptions . InvalidResponse , <EOL> exceptions . ResponseError ) ) , <EOL> ) <EOL> def get_redis_ConnectionError ( ) : <EOL> from redis import exceptions <EOL> return exceptions . ConnectionError <EOL> class MutexHeld ( Exception ) : <EOL> pass <EOL> @ contextmanager <EOL> def Mutex ( client , name , expire ) : <EOL> lock_id = uuid ( ) <EOL> i_won = client . setnx ( name , lock_id ) <EOL> try : <EOL> if i_won : <EOL> client . expire ( name , expire ) <EOL> yield <EOL> else : <EOL> if not client . ttl ( name ) : <EOL> client . expire ( name , expire ) <EOL> raise MutexHeld ( ) <EOL> finally : <EOL> if i_won : <EOL> try : <EOL> with client . pipeline ( True ) as pipe : <EOL> pipe . watch ( name ) <EOL> if pipe . get ( name ) == lock_id : <EOL> pipe . multi ( ) <EOL> pipe . delete ( name ) <EOL> pipe . execute ( ) <EOL> pipe . unwatch ( ) <EOL> except redis . WatchError : <EOL> pass <EOL> def _after_fork_cleanup_channel ( channel ) : <EOL> channel . _after_fork ( ) <EOL> class QoS ( virtual . QoS ) : <EOL> restore_at_shutdown = True <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( QoS , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _vrestore_count = <NUM_LIT:0> <EOL> def append ( self , message , delivery_tag ) : <EOL> delivery = message . delivery_info <EOL> EX , RK = delivery [ '<STR_LIT>' ] , delivery [ '<STR_LIT>' ] <EOL> with self . pipe_or_acquire ( ) as pipe : <EOL> pipe . zadd ( self . unacked_index_key , time ( ) , delivery_tag ) . hset ( self . unacked_key , delivery_tag , <EOL> dumps ( [ message . _raw , EX , RK ] ) ) . execute ( ) <EOL> super ( QoS , self ) . append ( message , delivery_tag ) <EOL> def restore_unacked ( self , client = None ) : <EOL> with self . channel . conn_or_acquire ( client ) as client : <EOL> for tag in self . _delivered : <EOL> self . restore_by_tag ( tag , client = client ) <EOL> self . _delivered . clear ( ) <EOL> def ack ( self , delivery_tag ) : <EOL> self . _remove_from_indices ( delivery_tag ) . execute ( ) <EOL> super ( QoS , self ) . ack ( delivery_tag ) <EOL> def reject ( self , delivery_tag , requeue = False ) : <EOL> if requeue : <EOL> self . restore_by_tag ( delivery_tag , leftmost = True ) <EOL> self . ack ( delivery_tag ) <EOL> @ contextmanager <EOL> def pipe_or_acquire ( self , pipe = None , client = None ) : <EOL> if pipe : <EOL> yield pipe <EOL> else : <EOL> with self . channel . conn_or_acquire ( client ) as client : <EOL> yield client . pipeline ( ) <EOL> def _remove_from_indices ( self , delivery_tag , pipe = None ) : <EOL> with self . pipe_or_acquire ( pipe ) as pipe : <EOL> return pipe . zrem ( self . unacked_index_key , delivery_tag ) . hdel ( self . unacked_key , delivery_tag ) <EOL> def restore_visible ( self , start = <NUM_LIT:0> , num = <NUM_LIT:10> , interval = <NUM_LIT:10> ) : <EOL> self . _vrestore_count += <NUM_LIT:1> <EOL> if ( self . _vrestore_count - <NUM_LIT:1> ) % interval : <EOL> return <EOL> with self . channel . conn_or_acquire ( ) as client : <EOL> ceil = time ( ) - self . visibility_timeout <EOL> try : <EOL> with Mutex ( client , self . unacked_mutex_key , <EOL> self . unacked_mutex_expire ) : <EOL> visible = client . zrevrangebyscore ( <EOL> self . unacked_index_key , ceil , <NUM_LIT:0> , <EOL> start = num and start , num = num , withscores = True ) <EOL> for tag , score in visible or [ ] : <EOL> self . restore_by_tag ( tag , client ) <EOL> except MutexHeld : <EOL> pass <EOL> def restore_by_tag ( self , tag , client = None , leftmost = False ) : <EOL> with self . channel . conn_or_acquire ( client ) as client : <EOL> with client . pipeline ( ) as pipe : <EOL> p , _ , _ = self . _remove_from_indices ( <EOL> tag , pipe . hget ( self . unacked_key , tag ) ) . execute ( ) <EOL> if p : <EOL> M , EX , RK = loads ( bytes_to_str ( p ) ) <EOL> self . channel . _do_restore_message ( M , EX , RK , client , leftmost ) <EOL> @ cached_property <EOL> def unacked_key ( self ) : <EOL> return self . channel . unacked_key <EOL> @ cached_property <EOL> def unacked_index_key ( self ) : <EOL> return self . channel . unacked_index_key <EOL> @ cached_property <EOL> def unacked_mutex_key ( self ) : <EOL> return self . channel . unacked_mutex_key <EOL> @ cached_property <EOL> def unacked_mutex_expire ( self ) : <EOL> return self . channel . unacked_mutex_expire <EOL> @ cached_property <EOL> def visibility_timeout ( self ) : <EOL> return self . channel . visibility_timeout <EOL> class MultiChannelPoller ( object ) : <EOL> eventflags = READ | ERR <EOL> _in_protected_read = False <EOL> after_read = None <EOL> def __init__ ( self ) : <EOL> self . _channels = set ( ) <EOL> self . _fd_to_chan = { } <EOL> self . _chan_to_sock = { } <EOL> self . poller = poll ( ) <EOL> self . after_read = set ( ) <EOL> def close ( self ) : <EOL> for fd in values ( self . _chan_to_sock ) : <EOL> try : <EOL> self . poller . unregister ( fd ) <EOL> except ( KeyError , ValueError ) : <EOL> pass <EOL> self . _channels . clear ( ) <EOL> self . _fd_to_chan . clear ( ) <EOL> self . _chan_to_sock . clear ( ) <EOL> def add ( self , channel ) : <EOL> self . _channels . add ( channel ) <EOL> def discard ( self , channel ) : <EOL> self . _channels . discard ( channel ) <EOL> def _on_connection_disconnect ( self , connection ) : <EOL> try : <EOL> self . poller . unregister ( connection . _sock ) <EOL> except ( AttributeError , TypeError ) : <EOL> pass <EOL> def _register ( self , channel , client , type ) : <EOL> if ( channel , client , type ) in self . _chan_to_sock : <EOL> self . _unregister ( channel , client , type ) <EOL> if client . connection . _sock is None : <EOL> client . connection . connect ( ) <EOL> sock = client . connection . _sock <EOL> self . _fd_to_chan [ sock . fileno ( ) ] = ( channel , type ) <EOL> self . _chan_to_sock [ ( channel , client , type ) ] = sock <EOL> self . poller . register ( sock , self . eventflags ) <EOL> def _unregister ( self , channel , client , type ) : <EOL> self . poller . unregister ( self . _chan_to_sock [ ( channel , client , type ) ] ) <EOL> def _client_registered ( self , channel , client , cmd ) : <EOL> return ( client . connection . _sock is not None and <EOL> ( channel , client , cmd ) in self . _chan_to_sock ) <EOL> def _register_BRPOP ( self , channel ) : <EOL> """<STR_LIT>""" <EOL> ident = channel , channel . client , '<STR_LIT>' <EOL> if not self . _client_registered ( channel , channel . client , '<STR_LIT>' ) : <EOL> channel . _in_poll = False <EOL> self . _register ( * ident ) <EOL> if not channel . _in_poll : <EOL> channel . _brpop_start ( ) <EOL> def _register_LISTEN ( self , channel ) : <EOL> """<STR_LIT>""" <EOL> if not self . _client_registered ( channel , channel . subclient , '<STR_LIT>' ) : <EOL> channel . _in_listen = False <EOL> self . _register ( channel , channel . subclient , '<STR_LIT>' ) <EOL> if not channel . _in_listen : <EOL> channel . _subscribe ( ) <EOL> def on_poll_start ( self ) : <EOL> for channel in self . _channels : <EOL> if channel . active_queues : <EOL> if channel . qos . can_consume ( ) : <EOL> self . _register_BRPOP ( channel ) <EOL> if channel . active_fanout_queues : <EOL> self . _register_LISTEN ( channel ) <EOL> def on_poll_init ( self , poller ) : <EOL> self . poller = poller <EOL> for channel in self . _channels : <EOL> return channel . qos . restore_visible ( <EOL> num = channel . unacked_restore_limit , <EOL> ) <EOL> def maybe_restore_messages ( self ) : <EOL> for channel in self . _channels : <EOL> if channel . active_queues : <EOL> return channel . qos . restore_visible ( <EOL> num = channel . unacked_restore_limit , <EOL> ) <EOL> def on_readable ( self , fileno ) : <EOL> chan , type = self . _fd_to_chan [ fileno ] <EOL> if chan . qos . can_consume ( ) : <EOL> return chan . handlers [ type ] ( ) <EOL> def handle_event ( self , fileno , event ) : <EOL> if event & READ : <EOL> return self . on_readable ( fileno ) , self <EOL> elif event & ERR : <EOL> chan , type = self . _fd_to_chan [ fileno ] <EOL> chan . _poll_error ( type ) <EOL> def get ( self , timeout = None ) : <EOL> self . _in_protected_read = True <EOL> try : <EOL> for channel in self . _channels : <EOL> if channel . active_queues : <EOL> if channel . qos . can_consume ( ) : <EOL> self . _register_BRPOP ( channel ) <EOL> if channel . active_fanout_queues : <EOL> self . _register_LISTEN ( channel ) <EOL> events = self . poller . poll ( timeout ) <EOL> for fileno , event in events or [ ] : <EOL> ret = self . handle_event ( fileno , event ) <EOL> if ret : <EOL> return ret <EOL> self . maybe_restore_messages ( ) <EOL> raise Empty ( ) <EOL> finally : <EOL> self . _in_protected_read = False <EOL> while self . after_read : <EOL> try : <EOL> fun = self . after_read . pop ( ) <EOL> except KeyError : <EOL> break <EOL> else : <EOL> fun ( ) <EOL> @ property <EOL> def fds ( self ) : <EOL> return self . _fd_to_chan <EOL> class Channel ( virtual . Channel ) : <EOL> QoS = QoS <EOL> _client = None <EOL> _subclient = None <EOL> _closing = False <EOL> supports_fanout = True <EOL> keyprefix_queue = '<STR_LIT>' <EOL> keyprefix_fanout = '<STR_LIT>' <EOL> sep = '<STR_LIT>' <EOL> _in_poll = False <EOL> _in_listen = False <EOL> _fanout_queues = { } <EOL> ack_emulation = True <EOL> unacked_key = '<STR_LIT>' <EOL> unacked_index_key = '<STR_LIT>' <EOL> unacked_mutex_key = '<STR_LIT>' <EOL> unacked_mutex_expire = <NUM_LIT> <EOL> unacked_restore_limit = None <EOL> visibility_timeout = <NUM_LIT> <EOL> priority_steps = PRIORITY_STEPS <EOL> socket_timeout = None <EOL> socket_connect_timeout = None <EOL> socket_keepalive = None <EOL> socket_keepalive_options = None <EOL> max_connections = <NUM_LIT:10> <EOL> fanout_prefix = True <EOL> fanout_patterns = True <EOL> queue_order_strategy = '<STR_LIT>' <EOL> _async_pool = None <EOL> _pool = None <EOL> from_transport_options = ( <EOL> virtual . Channel . from_transport_options + <EOL> ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> ) <EOL> connection_class = redis . Connection if redis else None <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super_ = super ( Channel , self ) <EOL> super_ . __init__ ( * args , ** kwargs ) <EOL> if not self . ack_emulation : <EOL> self . QoS = virtual . QoS <EOL> self . _queue_cycle = cycle_by_name ( self . queue_order_strategy ) ( ) <EOL> self . Client = self . _get_client ( ) <EOL> self . ResponseError = self . _get_response_error ( ) <EOL> self . active_fanout_queues = set ( ) <EOL> self . auto_delete_queues = set ( ) <EOL> self . _fanout_to_queue = { } <EOL> self . handlers = { '<STR_LIT>' : self . _brpop_read , '<STR_LIT>' : self . _receive } <EOL> if self . fanout_prefix : <EOL> if isinstance ( self . fanout_prefix , string_t ) : <EOL> self . keyprefix_fanout = self . fanout_prefix <EOL> else : <EOL> self . keyprefix_fanout = '<STR_LIT>' <EOL> try : <EOL> self . client . info ( ) <EOL> except Exception : <EOL> self . _disconnect_pools ( ) <EOL> raise <EOL> self . connection . cycle . add ( self ) <EOL> self . connection_errors = self . connection . connection_errors <EOL> if register_after_fork is not None : <EOL> register_after_fork ( self , _after_fork_cleanup_channel ) <EOL> def _after_fork ( self ) : <EOL> self . _disconnect_pools ( ) <EOL> def _disconnect_pools ( self ) : <EOL> pool = self . _pool <EOL> async_pool = self . _async_pool <EOL> self . _async_pool = self . _pool = None <EOL> if pool is not None : <EOL> pool . disconnect ( ) <EOL> if async_pool is not None : <EOL> async_pool . disconnect ( ) <EOL> def _on_connection_disconnect ( self , connection ) : <EOL> self . _in_poll = False <EOL> self . _in_listen = False <EOL> if self . connection and self . connection . cycle : <EOL> self . connection . cycle . _on_connection_disconnect ( connection ) <EOL> self . _disconnect_pools ( ) <EOL> if not self . _closing : <EOL> raise get_redis_ConnectionError ( ) <EOL> def _do_restore_message ( self , payload , exchange , routing_key , <EOL> client = None , leftmost = False ) : <EOL> with self . conn_or_acquire ( client ) as client : <EOL> try : <EOL> try : <EOL> payload [ '<STR_LIT>' ] [ '<STR_LIT>' ] = True <EOL> except KeyError : <EOL> pass <EOL> for queue in self . _lookup ( exchange , routing_key ) : <EOL> ( client . lpush if leftmost else client . rpush ) ( <EOL> queue , dumps ( payload ) , <EOL> ) <EOL> except Exception : <EOL> crit ( '<STR_LIT>' , payload , exc_info = True ) <EOL> def _restore ( self , message , leftmost = False ) : <EOL> if not self . ack_emulation : <EOL> return super ( Channel , self ) . _restore ( message ) <EOL> tag = message . delivery_tag <EOL> with self . conn_or_acquire ( ) as client : <EOL> with client . pipeline ( ) as pipe : <EOL> P , _ = pipe . hget ( self . unacked_key , tag ) . hdel ( self . unacked_key , tag ) . execute ( ) <EOL> if P : <EOL> M , EX , RK = loads ( bytes_to_str ( P ) ) <EOL> self . _do_restore_message ( M , EX , RK , client , leftmost ) <EOL> def _restore_at_beginning ( self , message ) : <EOL> return self . _restore ( message , leftmost = True ) <EOL> def basic_consume ( self , queue , * args , ** kwargs ) : <EOL> if queue in self . _fanout_queues : <EOL> exchange , _ = self . _fanout_queues [ queue ] <EOL> self . active_fanout_queues . add ( queue ) <EOL> self . _fanout_to_queue [ exchange ] = queue <EOL> ret = super ( Channel , self ) . basic_consume ( queue , * args , ** kwargs ) <EOL> self . _update_queue_cycle ( ) <EOL> return ret <EOL> def basic_cancel ( self , consumer_tag ) : <EOL> connection = self . connection <EOL> if connection : <EOL> if connection . cycle . _in_protected_read : <EOL> return connection . cycle . after_read . add ( <EOL> promise ( self . _basic_cancel , ( consumer_tag , ) ) , <EOL> ) <EOL> return self . _basic_cancel ( consumer_tag ) <EOL> def _basic_cancel ( self , consumer_tag ) : <EOL> try : <EOL> queue = self . _tag_to_queue [ consumer_tag ] <EOL> except KeyError : <EOL> return <EOL> try : <EOL> self . active_fanout_queues . remove ( queue ) <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> self . _unsubscribe_from ( queue ) <EOL> try : <EOL> exchange , _ = self . _fanout_queues [ queue ] <EOL> self . _fanout_to_queue . pop ( exchange ) <EOL> except KeyError : <EOL> pass <EOL> ret = super ( Channel , self ) . basic_cancel ( consumer_tag ) <EOL> self . _update_queue_cycle ( ) <EOL> return ret <EOL> def _get_publish_topic ( self , exchange , routing_key ) : <EOL> if routing_key and self . fanout_patterns : <EOL> return '<STR_LIT>' . join ( [ self . keyprefix_fanout , exchange , '<STR_LIT:/>' , routing_key ] ) <EOL> return '<STR_LIT>' . join ( [ self . keyprefix_fanout , exchange ] ) <EOL> def _get_subscribe_topic ( self , queue ) : <EOL> exchange , routing_key = self . _fanout_queues [ queue ] <EOL> return self . _get_publish_topic ( exchange , routing_key ) <EOL> def _subscribe ( self ) : <EOL> keys = [ self . _get_subscribe_topic ( queue ) <EOL> for queue in self . active_fanout_queues ] <EOL> if not keys : <EOL> return <EOL> c = self . subclient <EOL> if c . connection . _sock is None : <EOL> c . connection . connect ( ) <EOL> self . _in_listen = True <EOL> c . psubscribe ( keys ) <EOL> def _unsubscribe_from ( self , queue ) : <EOL> topic = self . _get_subscribe_topic ( queue ) <EOL> c = self . subclient <EOL> should_disconnect = False <EOL> if c . connection . _sock is None : <EOL> c . connection . connect ( ) <EOL> should_disconnect = True <EOL> try : <EOL> c . unsubscribe ( [ topic ] ) <EOL> finally : <EOL> if should_disconnect and c . connection : <EOL> c . connection . disconnect ( ) <EOL> def _handle_message ( self , client , r ) : <EOL> if bytes_to_str ( r [ <NUM_LIT:0> ] ) == '<STR_LIT>' and r [ <NUM_LIT:2> ] == <NUM_LIT:0> : <EOL> client . subscribed = False <EOL> elif bytes_to_str ( r [ <NUM_LIT:0> ] ) == '<STR_LIT>' : <EOL> return { '<STR_LIT:type>' : r [ <NUM_LIT:0> ] , '<STR_LIT>' : r [ <NUM_LIT:1> ] , <EOL> '<STR_LIT>' : r [ <NUM_LIT:2> ] , '<STR_LIT:data>' : r [ <NUM_LIT:3> ] } <EOL> else : <EOL> return { '<STR_LIT:type>' : r [ <NUM_LIT:0> ] , '<STR_LIT>' : None , <EOL> '<STR_LIT>' : r [ <NUM_LIT:1> ] , '<STR_LIT:data>' : r [ <NUM_LIT:2> ] } <EOL> def _receive ( self ) : <EOL> c = self . subclient <EOL> response = None <EOL> try : <EOL> response = c . parse_response ( ) <EOL> except self . connection_errors : <EOL> self . _in_listen = False <EOL> raise <EOL> if response is not None : <EOL> payload = self . _handle_message ( c , response ) <EOL> if bytes_to_str ( payload [ '<STR_LIT:type>' ] ) . endswith ( '<STR_LIT:message>' ) : <EOL> channel = bytes_to_str ( payload [ '<STR_LIT>' ] ) <EOL> if payload [ '<STR_LIT:data>' ] : <EOL> if channel [ <NUM_LIT:0> ] == '<STR_LIT:/>' : <EOL> _ , _ , channel = channel . partition ( '<STR_LIT:.>' ) <EOL> try : <EOL> message = loads ( bytes_to_str ( payload [ '<STR_LIT:data>' ] ) ) <EOL> except ( TypeError , ValueError ) : <EOL> warn ( '<STR_LIT>' , <EOL> channel , repr ( payload ) [ : <NUM_LIT> ] , exc_info = <NUM_LIT:1> ) <EOL> raise Empty ( ) <EOL> exchange = channel . split ( '<STR_LIT:/>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> return message , self . _fanout_to_queue [ exchange ] <EOL> raise Empty ( ) <EOL> def _brpop_start ( self , timeout = <NUM_LIT:1> ) : <EOL> queues = self . _queue_cycle . consume ( len ( self . active_queues ) ) <EOL> if not queues : <EOL> return <EOL> keys = [ self . _q_for_pri ( queue , pri ) for pri in self . priority_steps <EOL> for queue in queues ] + [ timeout or <NUM_LIT:0> ] <EOL> self . _in_poll = True <EOL> self . client . connection . send_command ( '<STR_LIT>' , * keys ) <EOL> def _brpop_read ( self , ** options ) : <EOL> try : <EOL> try : <EOL> dest__item = self . client . parse_response ( self . client . connection , <EOL> '<STR_LIT>' , <EOL> ** options ) <EOL> except self . connection_errors : <EOL> self . client . connection . disconnect ( ) <EOL> raise <EOL> if dest__item : <EOL> dest , item = dest__item <EOL> dest = bytes_to_str ( dest ) . rsplit ( self . sep , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> self . _queue_cycle . rotate ( dest ) <EOL> return loads ( bytes_to_str ( item ) ) , dest <EOL> else : <EOL> raise Empty ( ) <EOL> finally : <EOL> self . _in_poll = False <EOL> def _poll_error ( self , type , ** options ) : <EOL> if type == '<STR_LIT>' : <EOL> self . subclient . parse_response ( ) <EOL> else : <EOL> self . client . parse_response ( self . client . connection , type ) <EOL> def _get ( self , queue ) : <EOL> with self . conn_or_acquire ( ) as client : <EOL> for pri in self . priority_steps : <EOL> item = client . rpop ( self . _q_for_pri ( queue , pri ) ) <EOL> if item : <EOL> return loads ( bytes_to_str ( item ) ) <EOL> raise Empty ( ) <EOL> def _size ( self , queue ) : <EOL> with self . conn_or_acquire ( ) as client : <EOL> with client . pipeline ( ) as pipe : <EOL> for pri in self . priority_steps : <EOL> pipe = pipe . llen ( self . _q_for_pri ( queue , pri ) ) <EOL> sizes = pipe . execute ( ) <EOL> return sum ( size for size in sizes <EOL> if isinstance ( size , numbers . Integral ) ) <EOL> def _q_for_pri ( self , queue , pri ) : <EOL> pri = self . priority ( pri ) <EOL> return '<STR_LIT>' % ( ( queue , self . sep , pri ) if pri else ( queue , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> def priority ( self , n ) : <EOL> steps = self . priority_steps <EOL> return steps [ bisect ( steps , n ) - <NUM_LIT:1> ] <EOL> def _put ( self , queue , message , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> pri = self . _get_message_priority ( message , reverse = True ) <EOL> with self . conn_or_acquire ( ) as client : <EOL> client . lpush ( self . _q_for_pri ( queue , pri ) , dumps ( message ) ) <EOL> def _put_fanout ( self , exchange , message , routing_key , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> with self . conn_or_acquire ( ) as client : <EOL> client . publish ( <EOL> self . _get_publish_topic ( exchange , routing_key ) , <EOL> dumps ( message ) , <EOL> ) <EOL> def _new_queue ( self , queue , auto_delete = False , ** kwargs ) : <EOL> if auto_delete : <EOL> self . auto_delete_queues . add ( queue ) <EOL> def _queue_bind ( self , exchange , routing_key , pattern , queue ) : <EOL> if self . typeof ( exchange ) . type == '<STR_LIT>' : <EOL> self . _fanout_queues [ queue ] = ( <EOL> exchange , routing_key . replace ( '<STR_LIT:#>' , '<STR_LIT:*>' ) , <EOL> ) <EOL> with self . conn_or_acquire ( ) as client : <EOL> client . sadd ( self . keyprefix_queue % ( exchange , ) , <EOL> self . sep . join ( [ routing_key or '<STR_LIT>' , <EOL> pattern or '<STR_LIT>' , <EOL> queue or '<STR_LIT>' ] ) ) <EOL> def _delete ( self , queue , exchange , routing_key , pattern , * args , ** kwargs ) : <EOL> self . auto_delete_queues . discard ( queue ) <EOL> with self . conn_or_acquire ( client = kwargs . get ( '<STR_LIT>' ) ) as client : <EOL> client . srem ( self . keyprefix_queue % ( exchange , ) , <EOL> self . sep . join ( [ routing_key or '<STR_LIT>' , <EOL> pattern or '<STR_LIT>' , <EOL> queue or '<STR_LIT>' ] ) ) <EOL> with client . pipeline ( ) as pipe : <EOL> for pri in self . priority_steps : <EOL> pipe = pipe . delete ( self . _q_for_pri ( queue , pri ) ) <EOL> pipe . execute ( ) <EOL> def _has_queue ( self , queue , ** kwargs ) : <EOL> with self . conn_or_acquire ( ) as client : <EOL> with client . pipeline ( ) as pipe : <EOL> for pri in self . priority_steps : <EOL> pipe = pipe . exists ( self . _q_for_pri ( queue , pri ) ) <EOL> return any ( pipe . execute ( ) ) <EOL> def get_table ( self , exchange ) : <EOL> key = self . keyprefix_queue % exchange <EOL> with self . conn_or_acquire ( ) as client : <EOL> values = client . smembers ( key ) <EOL> if not values : <EOL> raise InconsistencyError ( NO_ROUTE_ERROR . format ( exchange , key ) ) <EOL> return [ tuple ( bytes_to_str ( val ) . split ( self . sep ) ) for val in values ] <EOL> def _purge ( self , queue ) : <EOL> with self . conn_or_acquire ( ) as client : <EOL> with client . pipeline ( ) as pipe : <EOL> for pri in self . priority_steps : <EOL> priq = self . _q_for_pri ( queue , pri ) <EOL> pipe = pipe . llen ( priq ) . delete ( priq ) <EOL> sizes = pipe . execute ( ) <EOL> return sum ( sizes [ : : <NUM_LIT:2> ] ) <EOL> def close ( self ) : <EOL> self . _closing = True <EOL> if not self . closed : <EOL> self . connection . cycle . discard ( self ) <EOL> client = self . __dict__ . get ( '<STR_LIT>' ) <EOL> if client is not None : <EOL> for queue in self . _fanout_queues : <EOL> if queue in self . auto_delete_queues : <EOL> self . queue_delete ( queue , client = client ) <EOL> self . _disconnect_pools ( ) <EOL> self . _close_clients ( ) <EOL> super ( Channel , self ) . close ( ) <EOL> def _close_clients ( self ) : <EOL> for attr in '<STR_LIT>' , '<STR_LIT>' : <EOL> try : <EOL> self . __dict__ [ attr ] . connection . disconnect ( ) <EOL> except ( KeyError , AttributeError , self . ResponseError ) : <EOL> pass <EOL> def _prepare_virtual_host ( self , vhost ) : <EOL> if not isinstance ( vhost , numbers . Integral ) : <EOL> if not vhost or vhost == '<STR_LIT:/>' : <EOL> vhost = DEFAULT_DB <EOL> elif vhost . startswith ( '<STR_LIT:/>' ) : <EOL> vhost = vhost [ <NUM_LIT:1> : ] <EOL> try : <EOL> vhost = int ( vhost ) <EOL> except ValueError : <EOL> raise ValueError ( <EOL> '<STR_LIT>' . format ( <EOL> vhost , <EOL> ) ) <EOL> return vhost <EOL> def _filter_tcp_connparams ( self , socket_keepalive = None , <EOL> socket_keepalive_options = None , ** params ) : <EOL> return params <EOL> def _connparams ( self , async = False ) : <EOL> conninfo = self . connection . client <EOL> connparams = { <EOL> '<STR_LIT:host>' : conninfo . hostname or '<STR_LIT:127.0.0.1>' , <EOL> '<STR_LIT:port>' : conninfo . port or self . connection . default_port , <EOL> '<STR_LIT>' : conninfo . virtual_host , <EOL> '<STR_LIT:password>' : conninfo . password , <EOL> '<STR_LIT>' : self . max_connections , <EOL> '<STR_LIT>' : self . socket_timeout , <EOL> '<STR_LIT>' : self . socket_connect_timeout , <EOL> '<STR_LIT>' : self . socket_keepalive , <EOL> '<STR_LIT>' : self . socket_keepalive_options , <EOL> } <EOL> if conninfo . ssl : <EOL> connparams [ '<STR_LIT>' ] = True <EOL> try : <EOL> connparams . update ( conninfo . ssl ) <EOL> except TypeError : <EOL> pass <EOL> host = connparams [ '<STR_LIT:host>' ] <EOL> if '<STR_LIT>' in host : <EOL> scheme , _ , _ , _ , _ , path , query = _parse_url ( host ) <EOL> if scheme == '<STR_LIT>' : <EOL> connparams = self . _filter_tcp_connparams ( ** connparams ) <EOL> connparams . update ( { <EOL> '<STR_LIT>' : redis . UnixDomainSocketConnection , <EOL> '<STR_LIT:path>' : '<STR_LIT:/>' + path } , ** query ) <EOL> connparams . pop ( '<STR_LIT:host>' , None ) <EOL> connparams . pop ( '<STR_LIT:port>' , None ) <EOL> connparams [ '<STR_LIT>' ] = self . _prepare_virtual_host ( <EOL> connparams . pop ( '<STR_LIT>' , None ) ) <EOL> channel = self <EOL> connection_cls = ( <EOL> connparams . get ( '<STR_LIT>' ) or <EOL> self . connection_class <EOL> ) <EOL> if async : <EOL> class Connection ( connection_cls ) : <EOL> def disconnect ( self ) : <EOL> super ( Connection , self ) . disconnect ( ) <EOL> channel . _on_connection_disconnect ( self ) <EOL> connection_cls = Connection <EOL> connparams [ '<STR_LIT>' ] = connection_cls <EOL> return connparams <EOL> def _create_client ( self , async = False ) : <EOL> if async : <EOL> return self . Client ( connection_pool = self . async_pool ) <EOL> return self . Client ( connection_pool = self . pool ) <EOL> def _get_pool ( self , async = False ) : <EOL> params = self . _connparams ( async = async ) <EOL> self . keyprefix_fanout = self . keyprefix_fanout . format ( db = params [ '<STR_LIT>' ] ) <EOL> return redis . ConnectionPool ( ** params ) <EOL> def _get_client ( self ) : <EOL> if redis . VERSION < ( <NUM_LIT:2> , <NUM_LIT:10> , <NUM_LIT:0> ) : <EOL> raise VersionMismatch ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( redis ) ) <EOL> class KombuRedis ( redis . StrictRedis ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( KombuRedis , self ) . __init__ ( * args , ** kwargs ) <EOL> self . connection = self . connection_pool . get_connection ( '<STR_LIT:_>' ) <EOL> return KombuRedis <EOL> @ contextmanager <EOL> def conn_or_acquire ( self , client = None ) : <EOL> if client : <EOL> yield client <EOL> else : <EOL> client = self . _create_client ( ) <EOL> try : <EOL> yield client <EOL> finally : <EOL> self . pool . release ( client . connection ) <EOL> @ property <EOL> def pool ( self ) : <EOL> if self . _pool is None : <EOL> self . _pool = self . _get_pool ( ) <EOL> return self . _pool <EOL> @ property <EOL> def async_pool ( self ) : <EOL> if self . _async_pool is None : <EOL> self . _async_pool = self . _get_pool ( async = True ) <EOL> return self . _async_pool <EOL> @ cached_property <EOL> def client ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _create_client ( async = True ) <EOL> @ cached_property <EOL> def subclient ( self ) : <EOL> """<STR_LIT>""" <EOL> client = self . _create_client ( async = True ) <EOL> pubsub = client . pubsub ( ) <EOL> pool = pubsub . connection_pool <EOL> pubsub . connection = pool . get_connection ( '<STR_LIT>' , pubsub . shard_hint ) <EOL> return pubsub <EOL> def _update_queue_cycle ( self ) : <EOL> self . _queue_cycle . update ( self . active_queues ) <EOL> def _get_response_error ( self ) : <EOL> from redis import exceptions <EOL> return exceptions . ResponseError <EOL> @ property <EOL> def active_queues ( self ) : <EOL> """<STR_LIT>""" <EOL> return { queue for queue in self . _active_queues <EOL> if queue not in self . active_fanout_queues } <EOL> class Transport ( virtual . Transport ) : <EOL> Channel = Channel <EOL> polling_interval = None <EOL> default_port = DEFAULT_PORT <EOL> driver_type = '<STR_LIT>' <EOL> driver_name = '<STR_LIT>' <EOL> implements = virtual . Transport . implements . extend ( <EOL> async = True , <EOL> exchange_type = frozenset ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> ) <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> if redis is None : <EOL> raise ImportError ( '<STR_LIT>' ) <EOL> super ( Transport , self ) . __init__ ( * args , ** kwargs ) <EOL> self . connection_errors , self . channel_errors = self . _get_errors ( ) <EOL> self . cycle = MultiChannelPoller ( ) <EOL> def driver_version ( self ) : <EOL> return redis . __version__ <EOL> def register_with_event_loop ( self , connection , loop ) : <EOL> cycle = self . cycle <EOL> cycle . on_poll_init ( loop . poller ) <EOL> cycle_poll_start = cycle . on_poll_start <EOL> add_reader = loop . add_reader <EOL> on_readable = self . on_readable <EOL> def _on_disconnect ( connection ) : <EOL> if connection . _sock : <EOL> loop . remove ( connection . _sock ) <EOL> cycle . _on_connection_disconnect = _on_disconnect <EOL> def on_poll_start ( ) : <EOL> cycle_poll_start ( ) <EOL> [ add_reader ( fd , on_readable , fd ) for fd in cycle . fds ] <EOL> loop . on_tick . add ( on_poll_start ) <EOL> loop . call_repeatedly ( <NUM_LIT:10> , cycle . maybe_restore_messages ) <EOL> def on_readable ( self , fileno ) : <EOL> """<STR_LIT>""" <EOL> item = self . cycle . on_readable ( fileno ) <EOL> if item : <EOL> self . _deliver ( * item ) <EOL> def _get_errors ( self ) : <EOL> """<STR_LIT>""" <EOL> return get_redis_error_classes ( ) <EOL> class SentinelChannel ( Channel ) : <EOL> """<STR_LIT>""" <EOL> from_transport_options = Channel . from_transport_options + ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> connection_class = sentinel . SentinelManagedConnection if sentinel else None <EOL> def _sentinel_managed_pool ( self , async = False ) : <EOL> connparams = self . _connparams ( async ) <EOL> additional_params = connparams . copy ( ) <EOL> del additional_params [ '<STR_LIT:host>' ] <EOL> del additional_params [ '<STR_LIT:port>' ] <EOL> sentinel_inst = sentinel . Sentinel ( <EOL> [ ( connparams [ '<STR_LIT:host>' ] , connparams [ '<STR_LIT:port>' ] ) ] , <EOL> min_other_sentinels = getattr ( self , '<STR_LIT>' , <NUM_LIT:0> ) , <EOL> sentinel_kwargs = getattr ( self , '<STR_LIT>' , { } ) , <EOL> ** additional_params <EOL> ) <EOL> master_name = getattr ( self , '<STR_LIT>' , None ) <EOL> return sentinel_inst . master_for ( <EOL> master_name , <EOL> self . Client , <EOL> ) . connection_pool <EOL> def _get_pool ( self , async = False ) : <EOL> return self . _sentinel_managed_pool ( async ) <EOL> class SentinelTransport ( Transport ) : <EOL> default_port = <NUM_LIT> <EOL> Channel = SentinelChannel </s>
