<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , print_function , unicode_literals <EOL> import base64 <EOL> import socket <EOL> import sys <EOL> import warnings <EOL> import collections <EOL> from array import array <EOL> from collections import OrderedDict <EOL> from itertools import count <EOL> from multiprocessing . util import Finalize <EOL> from time import sleep <EOL> from amqp . protocol import queue_declare_ok_t <EOL> from kombu . exceptions import ResourceError , ChannelError <EOL> from kombu . five import Empty , items , monotonic <EOL> from kombu . log import get_logger <EOL> from kombu . utils import emergency_dump_state , uuid <EOL> from kombu . utils . encoding import str_to_bytes , bytes_to_str <EOL> from kombu . utils . scheduling import FairCycle <EOL> from kombu . transport import base <EOL> from . exchange import STANDARD_EXCHANGE_TYPES <EOL> ARRAY_TYPE_H = '<STR_LIT:H>' if sys . version_info [ <NUM_LIT:0> ] == <NUM_LIT:3> else b'<STR_LIT:H>' <EOL> UNDELIVERABLE_FMT = """<STR_LIT>""" <EOL> NOT_EQUIVALENT_FMT = """<STR_LIT>""" <EOL> W_NO_CONSUMERS = """<STR_LIT>""" <EOL> RESTORING_FMT = '<STR_LIT>' <EOL> RESTORE_PANIC_FMT = '<STR_LIT>' <EOL> logger = get_logger ( __name__ ) <EOL> class Base64 ( object ) : <EOL> def encode ( self , s ) : <EOL> return bytes_to_str ( base64 . b64encode ( str_to_bytes ( s ) ) ) <EOL> def decode ( self , s ) : <EOL> return base64 . b64decode ( str_to_bytes ( s ) ) <EOL> class NotEquivalentError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class UndeliverableWarning ( UserWarning ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class BrokerState ( object ) : <EOL> exchanges = None <EOL> bindings = None <EOL> queue_index = None <EOL> def __init__ ( self , exchanges = None ) : <EOL> self . exchanges = { } if exchanges is None else exchanges <EOL> self . bindings = dict ( ) <EOL> self . queue_index = collections . defaultdict ( set ) <EOL> def clear ( self ) : <EOL> self . exchanges . clear ( ) <EOL> self . bindings . clear ( ) <EOL> self . queue_index . clear ( ) <EOL> def has_binding ( self , queue , exchange , routing_key ) : <EOL> return ( queue , exchange , routing_key ) in self . bindings <EOL> def binding_declare ( self , queue , exchange , routing_key , arguments ) : <EOL> if not self . has_binding ( queue , exchange , routing_key ) : <EOL> key = ( queue , exchange , routing_key ) <EOL> self . bindings [ key ] = arguments <EOL> self . queue_index [ queue ] . add ( key ) <EOL> def binding_delete ( self , queue , exchange , routing_key ) : <EOL> if self . has_binding ( queue , exchange , routing_key ) : <EOL> key = ( queue , exchange , routing_key ) <EOL> del self . bindings [ key ] <EOL> self . queue_index [ queue ] . remove ( key ) <EOL> def queue_bindings_delete ( self , queue ) : <EOL> if queue in self . queue_index : <EOL> for binding in self . queue_index [ queue ] : <EOL> del self . bindings [ binding ] <EOL> del self . queue_index [ queue ] <EOL> def queue_bindings ( self , queue ) : <EOL> for queue , exchange , routing_key in self . queue_index [ queue ] : <EOL> yield ( <EOL> exchange , <EOL> routing_key , <EOL> self . bindings [ ( queue , exchange , routing_key ) ] <EOL> ) <EOL> class QoS ( object ) : <EOL> """<STR_LIT>""" <EOL> prefetch_count = <NUM_LIT:0> <EOL> _delivered = None <EOL> _dirty = None <EOL> restore_at_shutdown = True <EOL> def __init__ ( self , channel , prefetch_count = <NUM_LIT:0> ) : <EOL> self . channel = channel <EOL> self . prefetch_count = prefetch_count or <NUM_LIT:0> <EOL> self . _delivered = OrderedDict ( ) <EOL> self . _delivered . restored = False <EOL> self . _dirty = set ( ) <EOL> self . _quick_ack = self . _dirty . add <EOL> self . _quick_append = self . _delivered . __setitem__ <EOL> self . _on_collect = Finalize ( <EOL> self , self . restore_unacked_once , exitpriority = <NUM_LIT:1> , <EOL> ) <EOL> def can_consume ( self ) : <EOL> """<STR_LIT>""" <EOL> pcount = self . prefetch_count <EOL> return not pcount or len ( self . _delivered ) - len ( self . _dirty ) < pcount <EOL> def can_consume_max_estimate ( self ) : <EOL> """<STR_LIT>""" <EOL> pcount = self . prefetch_count <EOL> if pcount : <EOL> return max ( pcount - ( len ( self . _delivered ) - len ( self . _dirty ) ) , <NUM_LIT:0> ) <EOL> def append ( self , message , delivery_tag ) : <EOL> """<STR_LIT>""" <EOL> if self . _dirty : <EOL> self . _flush ( ) <EOL> self . _quick_append ( delivery_tag , message ) <EOL> def get ( self , delivery_tag ) : <EOL> return self . _delivered [ delivery_tag ] <EOL> def _flush ( self ) : <EOL> """<STR_LIT>""" <EOL> dirty = self . _dirty <EOL> delivered = self . _delivered <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> dirty_tag = dirty . pop ( ) <EOL> except KeyError : <EOL> break <EOL> delivered . pop ( dirty_tag , None ) <EOL> def ack ( self , delivery_tag ) : <EOL> """<STR_LIT>""" <EOL> self . _quick_ack ( delivery_tag ) <EOL> def reject ( self , delivery_tag , requeue = False ) : <EOL> """<STR_LIT>""" <EOL> if requeue : <EOL> self . channel . _restore_at_beginning ( self . _delivered [ delivery_tag ] ) <EOL> self . _quick_ack ( delivery_tag ) <EOL> def restore_unacked ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _flush ( ) <EOL> delivered = self . _delivered <EOL> errors = [ ] <EOL> restore = self . channel . _restore <EOL> pop_message = delivered . popitem <EOL> while delivered : <EOL> try : <EOL> _ , message = pop_message ( ) <EOL> except KeyError : <EOL> break <EOL> try : <EOL> restore ( message ) <EOL> except BaseException as exc : <EOL> errors . append ( ( exc , message ) ) <EOL> delivered . clear ( ) <EOL> return errors <EOL> def restore_unacked_once ( self , stderr = None ) : <EOL> """<STR_LIT>""" <EOL> self . _on_collect . cancel ( ) <EOL> self . _flush ( ) <EOL> stderr = sys . stderr if stderr is None else stderr <EOL> state = self . _delivered <EOL> if not self . restore_at_shutdown or not self . channel . do_restore : <EOL> return <EOL> if getattr ( state , '<STR_LIT>' , None ) : <EOL> assert not state <EOL> return <EOL> try : <EOL> if state : <EOL> print ( RESTORING_FMT . format ( len ( self . _delivered ) ) , <EOL> file = stderr ) <EOL> unrestored = self . restore_unacked ( ) <EOL> if unrestored : <EOL> errors , messages = list ( zip ( * unrestored ) ) <EOL> print ( RESTORE_PANIC_FMT . format ( len ( errors ) , errors ) , <EOL> file = stderr ) <EOL> emergency_dump_state ( messages , stderr = stderr ) <EOL> finally : <EOL> state . restored = True <EOL> def restore_visible ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Message ( base . Message ) : <EOL> def __init__ ( self , channel , payload , ** kwargs ) : <EOL> self . _raw = payload <EOL> properties = payload [ '<STR_LIT>' ] <EOL> body = payload . get ( '<STR_LIT:body>' ) <EOL> if body : <EOL> body = channel . decode_body ( body , properties . get ( '<STR_LIT>' ) ) <EOL> kwargs . update ( { <EOL> '<STR_LIT:body>' : body , <EOL> '<STR_LIT>' : properties [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : payload . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : payload . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : payload . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : properties , <EOL> '<STR_LIT>' : properties . get ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : '<STR_LIT:utf-8>' , <EOL> } ) <EOL> super ( Message , self ) . __init__ ( channel , ** kwargs ) <EOL> def serializable ( self ) : <EOL> props = self . properties <EOL> body , _ = self . channel . encode_body ( self . body , <EOL> props . get ( '<STR_LIT>' ) ) <EOL> headers = dict ( self . headers ) <EOL> headers . pop ( '<STR_LIT>' , None ) <EOL> return { <EOL> '<STR_LIT:body>' : body , <EOL> '<STR_LIT>' : props , <EOL> '<STR_LIT>' : self . content_type , <EOL> '<STR_LIT>' : self . content_encoding , <EOL> '<STR_LIT>' : headers , <EOL> } <EOL> class AbstractChannel ( object ) : <EOL> """<STR_LIT>""" <EOL> def _get ( self , queue , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def _put ( self , queue , message ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def _purge ( self , queue ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def _size ( self , queue ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT:0> <EOL> def _delete ( self , queue , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . _purge ( queue ) <EOL> def _new_queue ( self , queue , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def _has_queue ( self , queue , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return True <EOL> def _poll ( self , cycle , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> return cycle . get ( ) <EOL> class Channel ( AbstractChannel , base . StdChannel ) : <EOL> """<STR_LIT>""" <EOL> Message = Message <EOL> QoS = QoS <EOL> do_restore = True <EOL> exchange_types = dict ( STANDARD_EXCHANGE_TYPES ) <EOL> supports_fanout = False <EOL> codecs = { '<STR_LIT>' : Base64 ( ) } <EOL> body_encoding = '<STR_LIT>' <EOL> _delivery_tags = count ( <NUM_LIT:1> ) <EOL> deadletter_queue = None <EOL> from_transport_options = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> default_priority = <NUM_LIT:0> <EOL> min_priority = <NUM_LIT:0> <EOL> max_priority = <NUM_LIT:9> <EOL> def __init__ ( self , connection , ** kwargs ) : <EOL> self . connection = connection <EOL> self . _consumers = set ( ) <EOL> self . _cycle = None <EOL> self . _tag_to_queue = { } <EOL> self . _active_queues = [ ] <EOL> self . _qos = None <EOL> self . closed = False <EOL> self . exchange_types = dict ( <EOL> ( typ , cls ( self ) ) for typ , cls in items ( self . exchange_types ) <EOL> ) <EOL> try : <EOL> self . channel_id = self . connection . _avail_channel_ids . pop ( ) <EOL> except IndexError : <EOL> raise ResourceError ( <EOL> '<STR_LIT>' . format ( <EOL> len ( self . connection . channels ) , <EOL> self . connection . channel_max ) , ( <NUM_LIT:20> , <NUM_LIT:10> ) , <EOL> ) <EOL> topts = self . connection . client . transport_options <EOL> for opt_name in self . from_transport_options : <EOL> try : <EOL> setattr ( self , opt_name , topts [ opt_name ] ) <EOL> except KeyError : <EOL> pass <EOL> def exchange_declare ( self , exchange = None , type = '<STR_LIT>' , durable = False , <EOL> auto_delete = False , arguments = None , <EOL> nowait = False , passive = False ) : <EOL> """<STR_LIT>""" <EOL> type = type or '<STR_LIT>' <EOL> exchange = exchange or '<STR_LIT>' % type <EOL> if passive : <EOL> if exchange not in self . state . exchanges : <EOL> raise ChannelError ( <EOL> '<STR_LIT>' . format ( <EOL> exchange , self . connection . client . virtual_host or '<STR_LIT:/>' ) , <EOL> ( <NUM_LIT:50> , <NUM_LIT:10> ) , '<STR_LIT>' , '<STR_LIT>' , <EOL> ) <EOL> return <EOL> try : <EOL> prev = self . state . exchanges [ exchange ] <EOL> if not self . typeof ( exchange ) . equivalent ( prev , exchange , type , <EOL> durable , auto_delete , <EOL> arguments ) : <EOL> raise NotEquivalentError ( NOT_EQUIVALENT_FMT . format ( <EOL> exchange , self . connection . client . virtual_host or '<STR_LIT:/>' ) ) <EOL> except KeyError : <EOL> self . state . exchanges [ exchange ] = { <EOL> '<STR_LIT:type>' : type , <EOL> '<STR_LIT>' : durable , <EOL> '<STR_LIT>' : auto_delete , <EOL> '<STR_LIT>' : arguments or { } , <EOL> '<STR_LIT>' : [ ] , <EOL> } <EOL> def exchange_delete ( self , exchange , if_unused = False , nowait = False ) : <EOL> """<STR_LIT>""" <EOL> for rkey , _ , queue in self . get_table ( exchange ) : <EOL> self . queue_delete ( queue , if_unused = True , if_empty = True ) <EOL> self . state . exchanges . pop ( exchange , None ) <EOL> def queue_declare ( self , queue = None , passive = False , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> queue = queue or '<STR_LIT>' % uuid ( ) <EOL> if passive and not self . _has_queue ( queue , ** kwargs ) : <EOL> raise ChannelError ( <EOL> '<STR_LIT>' . format ( <EOL> queue , self . connection . client . virtual_host or '<STR_LIT:/>' ) , <EOL> ( <NUM_LIT:50> , <NUM_LIT:10> ) , '<STR_LIT>' , '<STR_LIT>' , <EOL> ) <EOL> else : <EOL> self . _new_queue ( queue , ** kwargs ) <EOL> return queue_declare_ok_t ( queue , self . _size ( queue ) , <NUM_LIT:0> ) <EOL> def queue_delete ( self , queue , if_unused = False , if_empty = False , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if if_empty and self . _size ( queue ) : <EOL> return <EOL> for exchange , routing_key , args in self . state . queue_bindings ( queue ) : <EOL> meta = self . typeof ( exchange ) . prepare_bind ( <EOL> queue , exchange , routing_key , args , <EOL> ) <EOL> self . _delete ( queue , exchange , * meta , ** kwargs ) <EOL> self . state . queue_bindings_delete ( queue ) <EOL> def after_reply_message_received ( self , queue ) : <EOL> self . queue_delete ( queue ) <EOL> def exchange_bind ( self , destination , source = '<STR_LIT>' , routing_key = '<STR_LIT>' , <EOL> nowait = False , arguments = None ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def exchange_unbind ( self , destination , source = '<STR_LIT>' , routing_key = '<STR_LIT>' , <EOL> nowait = False , arguments = None ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def queue_bind ( self , queue , exchange = None , routing_key = '<STR_LIT>' , <EOL> arguments = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> exchange = exchange or '<STR_LIT>' <EOL> if self . state . has_binding ( queue , exchange , routing_key ) : <EOL> return <EOL> self . state . binding_declare ( queue , exchange , routing_key , arguments ) <EOL> table = self . state . exchanges [ exchange ] . setdefault ( '<STR_LIT>' , [ ] ) <EOL> meta = self . typeof ( exchange ) . prepare_bind ( <EOL> queue , exchange , routing_key , arguments , <EOL> ) <EOL> table . append ( meta ) <EOL> if self . supports_fanout : <EOL> self . _queue_bind ( exchange , * meta ) <EOL> def queue_unbind ( self , queue , exchange = None , routing_key = '<STR_LIT>' , <EOL> arguments = None , ** kwargs ) : <EOL> self . state . binding_delete ( queue , exchange , routing_key ) <EOL> try : <EOL> table = self . get_table ( exchange ) <EOL> except KeyError : <EOL> return <EOL> binding_meta = self . typeof ( exchange ) . prepare_bind ( <EOL> queue , exchange , routing_key , arguments , <EOL> ) <EOL> table [ : ] = [ meta for meta in table if meta != binding_meta ] <EOL> def list_bindings ( self ) : <EOL> return ( ( queue , exchange , rkey ) <EOL> for exchange in self . state . exchanges <EOL> for rkey , pattern , queue in self . get_table ( exchange ) ) <EOL> def queue_purge ( self , queue , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . _purge ( queue ) <EOL> def _next_delivery_tag ( self ) : <EOL> return uuid ( ) <EOL> def basic_publish ( self , message , exchange , routing_key , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> message [ '<STR_LIT:body>' ] , body_encoding = self . encode_body ( <EOL> message [ '<STR_LIT:body>' ] , self . body_encoding , <EOL> ) <EOL> props = message [ '<STR_LIT>' ] <EOL> props . update ( <EOL> body_encoding = body_encoding , <EOL> delivery_tag = self . _next_delivery_tag ( ) , <EOL> ) <EOL> props [ '<STR_LIT>' ] . update ( <EOL> exchange = exchange , <EOL> routing_key = routing_key , <EOL> ) <EOL> if exchange : <EOL> return self . typeof ( exchange ) . deliver ( <EOL> message , exchange , routing_key , ** kwargs <EOL> ) <EOL> return self . _put ( routing_key , message , ** kwargs ) <EOL> def basic_consume ( self , queue , no_ack , callback , consumer_tag , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . _tag_to_queue [ consumer_tag ] = queue <EOL> self . _active_queues . append ( queue ) <EOL> def _callback ( raw_message ) : <EOL> message = self . Message ( self , raw_message ) <EOL> if not no_ack : <EOL> self . qos . append ( message , message . delivery_tag ) <EOL> return callback ( message ) <EOL> self . connection . _callbacks [ queue ] = _callback <EOL> self . _consumers . add ( consumer_tag ) <EOL> self . _reset_cycle ( ) <EOL> def basic_cancel ( self , consumer_tag ) : <EOL> """<STR_LIT>""" <EOL> if consumer_tag in self . _consumers : <EOL> self . _consumers . remove ( consumer_tag ) <EOL> self . _reset_cycle ( ) <EOL> queue = self . _tag_to_queue . pop ( consumer_tag , None ) <EOL> try : <EOL> self . _active_queues . remove ( queue ) <EOL> except ValueError : <EOL> pass <EOL> self . connection . _callbacks . pop ( queue , None ) <EOL> def basic_get ( self , queue , no_ack = False , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> message = self . Message ( self , self . _get ( queue ) ) <EOL> if not no_ack : <EOL> self . qos . append ( message , message . delivery_tag ) <EOL> return message <EOL> except Empty : <EOL> pass <EOL> def basic_ack ( self , delivery_tag , multiple = False ) : <EOL> """<STR_LIT>""" <EOL> self . qos . ack ( delivery_tag ) <EOL> def basic_recover ( self , requeue = False ) : <EOL> """<STR_LIT>""" <EOL> if requeue : <EOL> return self . qos . restore_unacked ( ) <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def basic_reject ( self , delivery_tag , requeue = False ) : <EOL> """<STR_LIT>""" <EOL> self . qos . reject ( delivery_tag , requeue = requeue ) <EOL> def basic_qos ( self , prefetch_size = <NUM_LIT:0> , prefetch_count = <NUM_LIT:0> , <EOL> apply_global = False ) : <EOL> """<STR_LIT>""" <EOL> self . qos . prefetch_count = prefetch_count <EOL> def get_exchanges ( self ) : <EOL> return list ( self . state . exchanges ) <EOL> def get_table ( self , exchange ) : <EOL> """<STR_LIT>""" <EOL> return self . state . exchanges [ exchange ] [ '<STR_LIT>' ] <EOL> def typeof ( self , exchange , default = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> type = self . state . exchanges [ exchange ] [ '<STR_LIT:type>' ] <EOL> except KeyError : <EOL> type = default <EOL> return self . exchange_types [ type ] <EOL> def _lookup ( self , exchange , routing_key , default = None ) : <EOL> """<STR_LIT>""" <EOL> if default is None : <EOL> default = self . deadletter_queue <EOL> if not exchange : <EOL> return [ routing_key or default ] <EOL> try : <EOL> R = self . typeof ( exchange ) . lookup ( <EOL> self . get_table ( exchange ) , <EOL> exchange , routing_key , default , <EOL> ) <EOL> except KeyError : <EOL> R = [ ] <EOL> if not R and default is not None : <EOL> warnings . warn ( UndeliverableWarning ( UNDELIVERABLE_FMT . format ( <EOL> exchange = exchange , routing_key = routing_key ) ) , <EOL> ) <EOL> self . _new_queue ( default ) <EOL> R = [ default ] <EOL> return R <EOL> def _restore ( self , message ) : <EOL> """<STR_LIT>""" <EOL> delivery_info = message . delivery_info <EOL> message = message . serializable ( ) <EOL> message [ '<STR_LIT>' ] = True <EOL> for queue in self . _lookup ( <EOL> delivery_info [ '<STR_LIT>' ] , delivery_info [ '<STR_LIT>' ] ) : <EOL> self . _put ( queue , message ) <EOL> def _restore_at_beginning ( self , message ) : <EOL> return self . _restore ( message ) <EOL> def drain_events ( self , timeout = None ) : <EOL> if self . _consumers and self . qos . can_consume ( ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> return self . _get_many ( self . _active_queues , timeout = timeout ) <EOL> return self . _poll ( self . cycle , timeout = timeout ) <EOL> raise Empty ( ) <EOL> def message_to_python ( self , raw_message ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( raw_message , self . Message ) : <EOL> return self . Message ( self , payload = raw_message ) <EOL> return raw_message <EOL> def prepare_message ( self , body , priority = None , content_type = None , <EOL> content_encoding = None , headers = None , properties = None ) : <EOL> """<STR_LIT>""" <EOL> properties = properties or { } <EOL> properties . setdefault ( '<STR_LIT>' , { } ) <EOL> properties . setdefault ( '<STR_LIT>' , priority or self . default_priority ) <EOL> return { '<STR_LIT:body>' : body , <EOL> '<STR_LIT>' : content_encoding , <EOL> '<STR_LIT>' : content_type , <EOL> '<STR_LIT>' : headers or { } , <EOL> '<STR_LIT>' : properties or { } } <EOL> def flow ( self , active = True ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . closed : <EOL> self . closed = True <EOL> for consumer in list ( self . _consumers ) : <EOL> self . basic_cancel ( consumer ) <EOL> if self . _qos : <EOL> self . _qos . restore_unacked_once ( ) <EOL> if self . _cycle is not None : <EOL> self . _cycle . close ( ) <EOL> self . _cycle = None <EOL> if self . connection is not None : <EOL> self . connection . close_channel ( self ) <EOL> self . exchange_types = None <EOL> def encode_body ( self , body , encoding = None ) : <EOL> if encoding : <EOL> return self . codecs . get ( encoding ) . encode ( body ) , encoding <EOL> return body , encoding <EOL> def decode_body ( self , body , encoding = None ) : <EOL> if encoding : <EOL> return self . codecs . get ( encoding ) . decode ( body ) <EOL> return body <EOL> def _reset_cycle ( self ) : <EOL> self . _cycle = FairCycle ( self . _get , self . _active_queues , Empty ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , * exc_info ) : <EOL> self . close ( ) <EOL> @ property <EOL> def state ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . connection . state <EOL> @ property <EOL> def qos ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _qos is None : <EOL> self . _qos = self . QoS ( self ) <EOL> return self . _qos <EOL> @ property <EOL> def cycle ( self ) : <EOL> if self . _cycle is None : <EOL> self . _reset_cycle ( ) <EOL> return self . _cycle <EOL> def _get_message_priority ( self , message , reverse = False ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> priority = max ( <EOL> min ( int ( message [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) , <EOL> self . max_priority ) , <EOL> self . min_priority , <EOL> ) <EOL> except ( TypeError , ValueError , KeyError ) : <EOL> priority = self . default_priority <EOL> return ( self . max_priority - priority ) if reverse else priority <EOL> class Management ( base . Management ) : <EOL> def __init__ ( self , transport ) : <EOL> super ( Management , self ) . __init__ ( transport ) <EOL> self . channel = transport . client . channel ( ) <EOL> def get_bindings ( self ) : <EOL> return [ dict ( destination = q , source = e , routing_key = r ) <EOL> for q , e , r in self . channel . list_bindings ( ) ] <EOL> def close ( self ) : <EOL> self . channel . close ( ) <EOL> class Transport ( base . Transport ) : <EOL> """<STR_LIT>""" <EOL> Channel = Channel <EOL> Cycle = FairCycle <EOL> Management = Management <EOL> state = BrokerState ( ) <EOL> cycle = None <EOL> default_port = None <EOL> channels = None <EOL> _callbacks = None <EOL> polling_interval = <NUM_LIT:1.0> <EOL> channel_max = <NUM_LIT> <EOL> implements = base . Transport . implements . extend ( <EOL> async = False , <EOL> exchange_type = frozenset ( [ '<STR_LIT>' , '<STR_LIT>' ] ) , <EOL> heartbeats = False , <EOL> ) <EOL> def __init__ ( self , client , ** kwargs ) : <EOL> self . client = client <EOL> self . channels = [ ] <EOL> self . _avail_channels = [ ] <EOL> self . _callbacks = { } <EOL> self . cycle = self . Cycle ( self . _drain_channel , self . channels , Empty ) <EOL> polling_interval = client . transport_options . get ( '<STR_LIT>' ) <EOL> if polling_interval is not None : <EOL> self . polling_interval = polling_interval <EOL> self . _avail_channel_ids = array ( <EOL> ARRAY_TYPE_H , range ( self . channel_max , <NUM_LIT:0> , - <NUM_LIT:1> ) , <EOL> ) <EOL> def create_channel ( self , connection ) : <EOL> try : <EOL> return self . _avail_channels . pop ( ) <EOL> except IndexError : <EOL> channel = self . Channel ( connection ) <EOL> self . channels . append ( channel ) <EOL> return channel <EOL> def close_channel ( self , channel ) : <EOL> try : <EOL> self . _avail_channel_ids . append ( channel . channel_id ) <EOL> try : <EOL> self . channels . remove ( channel ) <EOL> except ValueError : <EOL> pass <EOL> finally : <EOL> channel . connection = None <EOL> def establish_connection ( self ) : <EOL> self . _avail_channels . append ( self . create_channel ( self ) ) <EOL> return self <EOL> def close_connection ( self , connection ) : <EOL> self . cycle . close ( ) <EOL> for l in self . _avail_channels , self . channels : <EOL> while l : <EOL> try : <EOL> channel = l . pop ( ) <EOL> except ( IndexError , KeyError ) : <EOL> pass <EOL> else : <EOL> channel . close ( ) <EOL> def drain_events ( self , connection , timeout = None ) : <EOL> loop = <NUM_LIT:0> <EOL> time_start = monotonic ( ) <EOL> get = self . cycle . get <EOL> polling_interval = self . polling_interval <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> item , channel = get ( timeout = timeout ) <EOL> except Empty : <EOL> if timeout and monotonic ( ) - time_start >= timeout : <EOL> raise socket . timeout ( ) <EOL> loop += <NUM_LIT:1> <EOL> if polling_interval is not None : <EOL> sleep ( polling_interval ) <EOL> else : <EOL> break <EOL> self . _deliver ( * item ) <EOL> def _deliver ( self , message , queue ) : <EOL> if not queue : <EOL> raise KeyError ( <EOL> '<STR_LIT>' . format ( <EOL> message ) ) <EOL> try : <EOL> callback = self . _callbacks [ queue ] <EOL> except KeyError : <EOL> logger . warn ( W_NO_CONSUMERS , queue ) <EOL> self . _reject_inbound_message ( message ) <EOL> else : <EOL> callback ( message ) <EOL> def _reject_inbound_message ( self , raw_message ) : <EOL> for channel in self . channels : <EOL> if channel : <EOL> message = channel . Message ( channel , raw_message ) <EOL> channel . qos . append ( message , message . delivery_tag ) <EOL> channel . basic_reject ( message . delivery_tag , requeue = True ) <EOL> break <EOL> def on_message_ready ( self , channel , message , queue ) : <EOL> if not queue or queue not in self . _callbacks : <EOL> raise KeyError ( <EOL> '<STR_LIT>' . format ( <EOL> queue , message ) ) <EOL> self . _callbacks [ queue ] ( message ) <EOL> def _drain_channel ( self , channel , timeout = None ) : <EOL> return channel . drain_events ( timeout = timeout ) <EOL> @ property <EOL> def default_connection_params ( self ) : <EOL> return { '<STR_LIT:port>' : self . default_port , '<STR_LIT>' : '<STR_LIT:localhost>' } </s>
