<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , unicode_literals <EOL> from itertools import count <EOL> from . common import maybe_declare <EOL> from . compression import compress <EOL> from . connection import maybe_channel , is_connection <EOL> from . entity import Exchange , Queue , maybe_delivery_mode <EOL> from . exceptions import ContentDisallowed <EOL> from . five import items , python_2_unicode_compatible , text_t , values <EOL> from . serialization import dumps , prepare_accept_content <EOL> from . utils import ChannelPromise , maybe_list <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> @ python_2_unicode_compatible <EOL> class Producer ( object ) : <EOL> """<STR_LIT>""" <EOL> exchange = None <EOL> routing_key = '<STR_LIT>' <EOL> serializer = None <EOL> compression = None <EOL> auto_declare = True <EOL> on_return = None <EOL> __connection__ = None <EOL> def __init__ ( self , channel , exchange = None , routing_key = None , <EOL> serializer = None , auto_declare = None , compression = None , <EOL> on_return = None ) : <EOL> self . _channel = channel <EOL> self . exchange = exchange <EOL> self . routing_key = routing_key or self . routing_key <EOL> self . serializer = serializer or self . serializer <EOL> self . compression = compression or self . compression <EOL> self . on_return = on_return or self . on_return <EOL> self . _channel_promise = None <EOL> if self . exchange is None : <EOL> self . exchange = Exchange ( '<STR_LIT>' ) <EOL> if auto_declare is not None : <EOL> self . auto_declare = auto_declare <EOL> if self . _channel : <EOL> self . revive ( self . _channel ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' . format ( self ) <EOL> def __reduce__ ( self ) : <EOL> return self . __class__ , self . __reduce_args__ ( ) <EOL> def __reduce_args__ ( self ) : <EOL> return ( None , self . exchange , self . routing_key , self . serializer , <EOL> self . auto_declare , self . compression ) <EOL> def declare ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . exchange . name : <EOL> self . exchange . declare ( ) <EOL> def maybe_declare ( self , entity , retry = False , ** retry_policy ) : <EOL> """<STR_LIT>""" <EOL> if entity : <EOL> return maybe_declare ( entity , self . channel , retry , ** retry_policy ) <EOL> def _delivery_details ( self , exchange , delivery_mode = None , <EOL> maybe_delivery_mode = maybe_delivery_mode , <EOL> Exchange = Exchange ) : <EOL> if isinstance ( exchange , Exchange ) : <EOL> return exchange . name , maybe_delivery_mode ( <EOL> delivery_mode or exchange . delivery_mode , <EOL> ) <EOL> return exchange , maybe_delivery_mode ( <EOL> delivery_mode or self . exchange . delivery_mode , <EOL> ) <EOL> def publish ( self , body , routing_key = None , delivery_mode = None , <EOL> mandatory = False , immediate = False , priority = <NUM_LIT:0> , <EOL> content_type = None , content_encoding = None , serializer = None , <EOL> headers = None , compression = None , exchange = None , retry = False , <EOL> retry_policy = None , declare = [ ] , expiration = None , ** properties ) : <EOL> """<STR_LIT>""" <EOL> _publish = self . _publish <EOL> headers = { } if headers is None else headers <EOL> retry_policy = { } if retry_policy is None else retry_policy <EOL> routing_key = self . routing_key if routing_key is None else routing_key <EOL> compression = self . compression if compression is None else compression <EOL> exchange_name , properties [ '<STR_LIT>' ] = self . _delivery_details ( <EOL> exchange or self . exchange , delivery_mode , <EOL> ) <EOL> if expiration is not None : <EOL> properties [ '<STR_LIT>' ] = str ( int ( expiration * <NUM_LIT:1000> ) ) <EOL> body , content_type , content_encoding = self . _prepare ( <EOL> body , serializer , content_type , content_encoding , <EOL> compression , headers ) <EOL> if retry : <EOL> _publish = self . connection . ensure ( self , _publish , ** retry_policy ) <EOL> return _publish ( <EOL> body , priority , content_type , content_encoding , <EOL> headers , properties , routing_key , mandatory , immediate , <EOL> exchange_name , declare , <EOL> ) <EOL> def _publish ( self , body , priority , content_type , content_encoding , <EOL> headers , properties , routing_key , mandatory , <EOL> immediate , exchange , declare ) : <EOL> channel = self . channel <EOL> message = channel . prepare_message ( <EOL> body , priority , content_type , <EOL> content_encoding , headers , properties , <EOL> ) <EOL> if declare : <EOL> maybe_declare = self . maybe_declare <EOL> [ maybe_declare ( entity ) for entity in declare ] <EOL> reply_to = properties . get ( '<STR_LIT>' ) <EOL> if isinstance ( reply_to , Queue ) : <EOL> properties [ '<STR_LIT>' ] = reply_to . name <EOL> return channel . basic_publish ( <EOL> message , <EOL> exchange = exchange , routing_key = routing_key , <EOL> mandatory = mandatory , immediate = immediate , <EOL> ) <EOL> def _get_channel ( self ) : <EOL> channel = self . _channel <EOL> if isinstance ( channel , ChannelPromise ) : <EOL> channel = self . _channel = channel ( ) <EOL> self . exchange . revive ( channel ) <EOL> if self . on_return : <EOL> channel . events [ '<STR_LIT>' ] . add ( self . on_return ) <EOL> return channel <EOL> def _set_channel ( self , channel ) : <EOL> self . _channel = channel <EOL> channel = property ( _get_channel , _set_channel ) <EOL> def revive ( self , channel ) : <EOL> """<STR_LIT>""" <EOL> if is_connection ( channel ) : <EOL> connection = channel <EOL> self . __connection__ = connection <EOL> channel = ChannelPromise ( lambda : connection . default_channel ) <EOL> if isinstance ( channel , ChannelPromise ) : <EOL> self . _channel = channel <EOL> self . exchange = self . exchange ( channel ) <EOL> else : <EOL> self . _channel = channel <EOL> if self . on_return : <EOL> self . _channel . events [ '<STR_LIT>' ] . add ( self . on_return ) <EOL> self . exchange = self . exchange ( channel ) <EOL> if self . auto_declare : <EOL> self . declare ( ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , * exc_info ) : <EOL> self . release ( ) <EOL> def release ( self ) : <EOL> pass <EOL> close = release <EOL> def _prepare ( self , body , serializer = None , content_type = None , <EOL> content_encoding = None , compression = None , headers = None ) : <EOL> if not content_type : <EOL> serializer = serializer or self . serializer <EOL> ( content_type , content_encoding , <EOL> body ) = dumps ( body , serializer = serializer ) <EOL> else : <EOL> if isinstance ( body , text_t ) : <EOL> if not content_encoding : <EOL> content_encoding = '<STR_LIT:utf-8>' <EOL> body = body . encode ( content_encoding ) <EOL> elif not content_encoding : <EOL> content_encoding = '<STR_LIT>' <EOL> if compression : <EOL> body , headers [ '<STR_LIT>' ] = compress ( body , compression ) <EOL> return body , content_type , content_encoding <EOL> @ property <EOL> def connection ( self ) : <EOL> try : <EOL> return self . __connection__ or self . channel . connection . client <EOL> except AttributeError : <EOL> pass <EOL> @ python_2_unicode_compatible <EOL> class Consumer ( object ) : <EOL> """<STR_LIT>""" <EOL> ContentDisallowed = ContentDisallowed <EOL> channel = None <EOL> queues = None <EOL> no_ack = None <EOL> auto_declare = True <EOL> callbacks = None <EOL> on_message = None <EOL> on_decode_error = None <EOL> accept = None <EOL> prefetch_count = None <EOL> _queues = None <EOL> _tags = count ( <NUM_LIT:1> ) <EOL> def __init__ ( self , channel , queues = None , no_ack = None , auto_declare = None , <EOL> callbacks = None , on_decode_error = None , on_message = None , <EOL> accept = None , prefetch_count = None , tag_prefix = None ) : <EOL> self . channel = channel <EOL> self . queues = maybe_list ( queues or [ ] ) <EOL> self . no_ack = self . no_ack if no_ack is None else no_ack <EOL> self . callbacks = ( self . callbacks or [ ] if callbacks is None <EOL> else callbacks ) <EOL> self . on_message = on_message <EOL> self . tag_prefix = tag_prefix <EOL> self . _active_tags = { } <EOL> if auto_declare is not None : <EOL> self . auto_declare = auto_declare <EOL> if on_decode_error is not None : <EOL> self . on_decode_error = on_decode_error <EOL> self . accept = prepare_accept_content ( accept ) <EOL> self . prefetch_count = prefetch_count <EOL> if self . channel : <EOL> self . revive ( self . channel ) <EOL> @ property <EOL> def queues ( self ) : <EOL> return list ( self . _queues . values ( ) ) <EOL> @ queues . setter <EOL> def queues ( self , queues ) : <EOL> self . _queues = { q . name : q for q in queues } <EOL> def revive ( self , channel ) : <EOL> """<STR_LIT>""" <EOL> self . _active_tags . clear ( ) <EOL> channel = self . channel = maybe_channel ( channel ) <EOL> for qname , queue in items ( self . _queues ) : <EOL> self . _queues . pop ( qname , None ) <EOL> queue = self . _queues [ queue . name ] = queue ( self . channel ) <EOL> queue . revive ( channel ) <EOL> if self . auto_declare : <EOL> self . declare ( ) <EOL> if self . prefetch_count is not None : <EOL> self . qos ( prefetch_count = self . prefetch_count ) <EOL> def declare ( self ) : <EOL> """<STR_LIT>""" <EOL> for queue in values ( self . _queues ) : <EOL> queue . declare ( ) <EOL> def register_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . callbacks . append ( callback ) <EOL> def __enter__ ( self ) : <EOL> self . consume ( ) <EOL> return self <EOL> def __exit__ ( self , * exc_info ) : <EOL> try : <EOL> self . cancel ( ) <EOL> except Exception : <EOL> pass <EOL> def add_queue ( self , queue ) : <EOL> """<STR_LIT>""" <EOL> queue = queue ( self . channel ) <EOL> if self . auto_declare : <EOL> queue . declare ( ) <EOL> self . _queues [ queue . name ] = queue <EOL> return queue <EOL> def consume ( self , no_ack = None ) : <EOL> """<STR_LIT>""" <EOL> queues = list ( values ( self . _queues ) ) <EOL> if queues : <EOL> no_ack = self . no_ack if no_ack is None else no_ack <EOL> H , T = queues [ : - <NUM_LIT:1> ] , queues [ - <NUM_LIT:1> ] <EOL> for queue in H : <EOL> self . _basic_consume ( queue , no_ack = no_ack , nowait = True ) <EOL> self . _basic_consume ( T , no_ack = no_ack , nowait = False ) <EOL> def cancel ( self ) : <EOL> """<STR_LIT>""" <EOL> cancel = self . channel . basic_cancel <EOL> for tag in values ( self . _active_tags ) : <EOL> cancel ( tag ) <EOL> self . _active_tags . clear ( ) <EOL> close = cancel <EOL> def cancel_by_queue ( self , queue ) : <EOL> """<STR_LIT>""" <EOL> qname = queue . name if isinstance ( queue , Queue ) else queue <EOL> try : <EOL> tag = self . _active_tags . pop ( qname ) <EOL> except KeyError : <EOL> pass <EOL> else : <EOL> self . channel . basic_cancel ( tag ) <EOL> finally : <EOL> self . _queues . pop ( qname , None ) <EOL> def consuming_from ( self , queue ) : <EOL> """<STR_LIT>""" <EOL> name = queue <EOL> if isinstance ( queue , Queue ) : <EOL> name = queue . name <EOL> return name in self . _active_tags <EOL> def purge ( self ) : <EOL> """<STR_LIT>""" <EOL> return sum ( queue . purge ( ) for queue in values ( self . _queues ) ) <EOL> def flow ( self , active ) : <EOL> """<STR_LIT>""" <EOL> self . channel . flow ( active ) <EOL> def qos ( self , prefetch_size = <NUM_LIT:0> , prefetch_count = <NUM_LIT:0> , apply_global = False ) : <EOL> """<STR_LIT>""" <EOL> return self . channel . basic_qos ( prefetch_size , <EOL> prefetch_count , <EOL> apply_global ) <EOL> def recover ( self , requeue = False ) : <EOL> """<STR_LIT>""" <EOL> return self . channel . basic_recover ( requeue = requeue ) <EOL> def receive ( self , body , message ) : <EOL> """<STR_LIT>""" <EOL> callbacks = self . callbacks <EOL> if not callbacks : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> [ callback ( body , message ) for callback in callbacks ] <EOL> def _basic_consume ( self , queue , consumer_tag = None , <EOL> no_ack = no_ack , nowait = True ) : <EOL> tag = self . _active_tags . get ( queue . name ) <EOL> if tag is None : <EOL> tag = self . _add_tag ( queue , consumer_tag ) <EOL> queue . consume ( tag , self . _receive_callback , <EOL> no_ack = no_ack , nowait = nowait ) <EOL> return tag <EOL> def _add_tag ( self , queue , consumer_tag = None ) : <EOL> tag = consumer_tag or '<STR_LIT>' . format ( <EOL> self . tag_prefix , next ( self . _tags ) ) <EOL> self . _active_tags [ queue . name ] = tag <EOL> return tag <EOL> def _receive_callback ( self , message ) : <EOL> accept = self . accept <EOL> on_m , channel , decoded = self . on_message , self . channel , None <EOL> try : <EOL> m2p = getattr ( channel , '<STR_LIT>' , None ) <EOL> if m2p : <EOL> message = m2p ( message ) <EOL> if accept is not None : <EOL> message . accept = accept <EOL> if message . errors : <EOL> return message . _reraise_error ( self . on_decode_error ) <EOL> decoded = None if on_m else message . decode ( ) <EOL> except Exception as exc : <EOL> if not self . on_decode_error : <EOL> raise <EOL> self . on_decode_error ( message , exc ) <EOL> else : <EOL> return on_m ( message ) if on_m else self . receive ( decoded , message ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' . format ( self ) <EOL> @ property <EOL> def connection ( self ) : <EOL> try : <EOL> return self . channel . connection . client <EOL> except AttributeError : <EOL> pass </s>
