<s> """<STR_LIT>""" <EOL> import collections <EOL> import logging <EOL> import uuid <EOL> import pika . frame as frame <EOL> import pika . exceptions as exceptions <EOL> import pika . spec as spec <EOL> from pika . utils import is_callable <EOL> from pika . compat import unicode_type , dictkeys , is_integer <EOL> LOGGER = logging . getLogger ( __name__ ) <EOL> MAX_CHANNELS = <NUM_LIT> <EOL> class Channel ( object ) : <EOL> """<STR_LIT>""" <EOL> CLOSED = <NUM_LIT:0> <EOL> OPENING = <NUM_LIT:1> <EOL> OPEN = <NUM_LIT:2> <EOL> CLOSING = <NUM_LIT:3> <EOL> _STATE_NAMES = { <EOL> CLOSED : '<STR_LIT>' , <EOL> OPENING : '<STR_LIT>' , <EOL> OPEN : '<STR_LIT>' , <EOL> CLOSING : '<STR_LIT>' <EOL> } <EOL> _ON_CHANNEL_CLEANUP_CB_KEY = '<STR_LIT>' <EOL> def __init__ ( self , connection , channel_number , on_open_callback ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( channel_number , int ) : <EOL> raise exceptions . InvalidChannelNumber <EOL> self . channel_number = channel_number <EOL> self . callbacks = connection . callbacks <EOL> self . connection = connection <EOL> self . flow_active = True <EOL> self . _content_assembler = ContentFrameAssembler ( ) <EOL> self . _blocked = collections . deque ( list ( ) ) <EOL> self . _blocking = None <EOL> self . _has_on_flow_callback = False <EOL> self . _cancelled = set ( ) <EOL> self . _consumers = dict ( ) <EOL> self . _consumers_with_noack = set ( ) <EOL> self . _on_flowok_callback = None <EOL> self . _on_getok_callback = None <EOL> self . _on_openok_callback = on_open_callback <EOL> self . _state = self . CLOSED <EOL> self . _closing_code_and_text = ( <NUM_LIT:0> , '<STR_LIT>' ) <EOL> self . _cookie = None <EOL> def __int__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . channel_number <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , <EOL> self . channel_number , <EOL> self . _STATE_NAMES [ self . _state ] , <EOL> self . connection ) <EOL> def add_callback ( self , callback , replies , one_shot = True ) : <EOL> """<STR_LIT>""" <EOL> for reply in replies : <EOL> self . callbacks . add ( self . channel_number , reply , callback , one_shot ) <EOL> def add_on_cancel_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . callbacks . add ( self . channel_number , spec . Basic . Cancel , callback , <EOL> False ) <EOL> def add_on_close_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . callbacks . add ( self . channel_number , '<STR_LIT>' , callback , <EOL> False , self ) <EOL> def add_on_flow_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . _has_on_flow_callback = True <EOL> self . callbacks . add ( self . channel_number , spec . Channel . Flow , callback , <EOL> False ) <EOL> def add_on_return_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . callbacks . add ( self . channel_number , '<STR_LIT>' , callback , False ) <EOL> def basic_ack ( self , delivery_tag = <NUM_LIT:0> , multiple = False ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise exceptions . ChannelClosed ( ) <EOL> return self . _send_method ( spec . Basic . Ack ( delivery_tag , multiple ) ) <EOL> def basic_cancel ( self , callback = None , consumer_tag = '<STR_LIT>' , nowait = False ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> if nowait : <EOL> if callback is not None : <EOL> raise ValueError ( <EOL> '<STR_LIT>' ) <EOL> else : <EOL> if callback is None : <EOL> raise ValueError ( <EOL> '<STR_LIT>' ) <EOL> if consumer_tag in self . _cancelled : <EOL> LOGGER . warning ( '<STR_LIT>' , <EOL> consumer_tag ) <EOL> return <EOL> if consumer_tag not in self . _consumers : <EOL> LOGGER . warning ( '<STR_LIT>' , <EOL> consumer_tag ) <EOL> return <EOL> LOGGER . debug ( '<STR_LIT>' , <EOL> consumer_tag , nowait ) <EOL> if nowait : <EOL> del self . _consumers [ consumer_tag ] <EOL> if callback is not None : <EOL> if nowait : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . callbacks . add ( self . channel_number , spec . Basic . CancelOk , <EOL> callback ) <EOL> self . _cancelled . add ( consumer_tag ) <EOL> self . _rpc ( spec . Basic . Cancel ( consumer_tag = consumer_tag , nowait = nowait ) , <EOL> self . _on_cancelok if not nowait else None , <EOL> [ ( spec . Basic . CancelOk , { '<STR_LIT>' : consumer_tag } ) ] if <EOL> nowait is False else [ ] ) <EOL> def basic_consume ( self , consumer_callback , <EOL> queue = '<STR_LIT>' , <EOL> no_ack = False , <EOL> exclusive = False , <EOL> consumer_tag = None , <EOL> arguments = None ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( consumer_callback ) <EOL> if not consumer_tag : <EOL> consumer_tag = self . _generate_consumer_tag ( ) <EOL> if consumer_tag in self . _consumers or consumer_tag in self . _cancelled : <EOL> raise exceptions . DuplicateConsumerTag ( consumer_tag ) <EOL> if no_ack : <EOL> self . _consumers_with_noack . add ( consumer_tag ) <EOL> self . _consumers [ consumer_tag ] = consumer_callback <EOL> self . _rpc ( spec . Basic . Consume ( queue = queue , <EOL> consumer_tag = consumer_tag , <EOL> no_ack = no_ack , <EOL> exclusive = exclusive , <EOL> arguments = arguments or dict ( ) ) , <EOL> self . _on_eventok , [ ( spec . Basic . ConsumeOk , <EOL> { '<STR_LIT>' : consumer_tag } ) ] ) <EOL> return consumer_tag <EOL> def _generate_consumer_tag ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( self . channel_number , <EOL> uuid . uuid4 ( ) . hex ) <EOL> def basic_get ( self , callback = None , queue = '<STR_LIT>' , no_ack = False ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> if self . _on_getok_callback is not None : <EOL> raise exceptions . DuplicateGetOkCallback ( ) <EOL> self . _on_getok_callback = callback <EOL> self . _send_method ( spec . Basic . Get ( queue = queue , no_ack = no_ack ) ) <EOL> def basic_nack ( self , delivery_tag = None , multiple = False , requeue = True ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise exceptions . ChannelClosed ( ) <EOL> return self . _send_method ( spec . Basic . Nack ( delivery_tag , multiple , <EOL> requeue ) ) <EOL> def basic_publish ( self , exchange , routing_key , body , <EOL> properties = None , <EOL> mandatory = False , <EOL> immediate = False ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise exceptions . ChannelClosed ( ) <EOL> if immediate : <EOL> LOGGER . warning ( '<STR_LIT>' ) <EOL> if isinstance ( body , unicode_type ) : <EOL> body = body . encode ( '<STR_LIT:utf-8>' ) <EOL> properties = properties or spec . BasicProperties ( ) <EOL> self . _send_method ( spec . Basic . Publish ( exchange = exchange , <EOL> routing_key = routing_key , <EOL> mandatory = mandatory , <EOL> immediate = immediate ) , <EOL> ( properties , body ) ) <EOL> def basic_qos ( self , <EOL> callback = None , <EOL> prefetch_size = <NUM_LIT:0> , <EOL> prefetch_count = <NUM_LIT:0> , <EOL> all_channels = False ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Basic . Qos ( prefetch_size , prefetch_count , <EOL> all_channels ) , <EOL> callback , [ spec . Basic . QosOk ] ) <EOL> def basic_reject ( self , delivery_tag , requeue = True ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise exceptions . ChannelClosed ( ) <EOL> if not is_integer ( delivery_tag ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> return self . _send_method ( spec . Basic . Reject ( delivery_tag , requeue ) ) <EOL> def basic_recover ( self , callback = None , requeue = False ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Basic . Recover ( requeue ) , callback , <EOL> [ spec . Basic . RecoverOk ] ) <EOL> def close ( self , reply_code = <NUM_LIT:0> , reply_text = "<STR_LIT>" ) : <EOL> """<STR_LIT>""" <EOL> if self . is_closed : <EOL> raise exceptions . ChannelClosed ( '<STR_LIT>' % self ) <EOL> if self . is_closing : <EOL> raise exceptions . ChannelAlreadyClosing ( '<STR_LIT>' % self ) <EOL> LOGGER . info ( '<STR_LIT>' , <EOL> reply_code , reply_text , self ) <EOL> for consumer_tag in dictkeys ( self . _consumers ) : <EOL> if consumer_tag not in self . _cancelled : <EOL> self . basic_cancel ( consumer_tag = consumer_tag ) <EOL> self . _set_state ( self . CLOSING ) <EOL> self . _rpc ( spec . Channel . Close ( reply_code , reply_text , <NUM_LIT:0> , <NUM_LIT:0> ) , <EOL> self . _on_closeok , [ spec . Channel . CloseOk ] ) <EOL> def confirm_delivery ( self , callback = None , nowait = False ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> if not ( self . connection . publisher_confirms and <EOL> self . connection . basic_nack ) : <EOL> raise exceptions . MethodNotImplemented ( '<STR_LIT>' ) <EOL> if callback is not None : <EOL> self . callbacks . add ( self . channel_number , spec . Basic . Ack , callback , <EOL> False ) <EOL> self . callbacks . add ( self . channel_number , spec . Basic . Nack , callback , <EOL> False ) <EOL> self . _rpc ( spec . Confirm . Select ( nowait ) , <EOL> self . _on_selectok if not nowait else None , <EOL> [ spec . Confirm . SelectOk ] if nowait is False else [ ] ) <EOL> @ property <EOL> def consumer_tags ( self ) : <EOL> """<STR_LIT>""" <EOL> return dictkeys ( self . _consumers ) <EOL> def exchange_bind ( self , <EOL> callback = None , <EOL> destination = None , <EOL> source = None , <EOL> routing_key = '<STR_LIT>' , <EOL> nowait = False , <EOL> arguments = None ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Exchange . Bind ( <NUM_LIT:0> , destination , source , routing_key , <EOL> nowait , arguments or dict ( ) ) , <EOL> callback , [ spec . Exchange . BindOk ] if nowait is False <EOL> else [ ] ) <EOL> def exchange_declare ( self , <EOL> callback = None , <EOL> exchange = None , <EOL> exchange_type = '<STR_LIT>' , <EOL> passive = False , <EOL> durable = False , <EOL> auto_delete = False , <EOL> internal = False , <EOL> nowait = False , <EOL> arguments = None ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Exchange . Declare ( <NUM_LIT:0> , exchange , exchange_type , <EOL> passive , durable , auto_delete , <EOL> internal , nowait , <EOL> arguments or dict ( ) ) , <EOL> callback , <EOL> [ spec . Exchange . DeclareOk ] if nowait is False else [ ] ) <EOL> def exchange_delete ( self , <EOL> callback = None , <EOL> exchange = None , <EOL> if_unused = False , <EOL> nowait = False ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Exchange . Delete ( <NUM_LIT:0> , exchange , if_unused , nowait ) , <EOL> callback , [ spec . Exchange . DeleteOk ] if nowait is False <EOL> else [ ] ) <EOL> def exchange_unbind ( self , <EOL> callback = None , <EOL> destination = None , <EOL> source = None , <EOL> routing_key = '<STR_LIT>' , <EOL> nowait = False , <EOL> arguments = None ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Exchange . Unbind ( <NUM_LIT:0> , destination , source , <EOL> routing_key , nowait , arguments ) , <EOL> callback , <EOL> [ spec . Exchange . UnbindOk ] if nowait is False else [ ] ) <EOL> def flow ( self , callback , active ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> self . _on_flowok_callback = callback <EOL> self . _rpc ( spec . Channel . Flow ( active ) , self . _on_flowok , <EOL> [ spec . Channel . FlowOk ] ) <EOL> @ property <EOL> def is_closed ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _state == self . CLOSED <EOL> @ property <EOL> def is_closing ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _state == self . CLOSING <EOL> @ property <EOL> def is_open ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _state == self . OPEN <EOL> def open ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _set_state ( self . OPENING ) <EOL> self . _add_callbacks ( ) <EOL> self . _rpc ( spec . Channel . Open ( ) , self . _on_openok , [ spec . Channel . OpenOk ] ) <EOL> def queue_bind ( self , callback , queue , exchange , <EOL> routing_key = None , <EOL> nowait = False , <EOL> arguments = None ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> replies = [ spec . Queue . BindOk ] if nowait is False else [ ] <EOL> if routing_key is None : <EOL> routing_key = queue <EOL> return self . _rpc ( spec . Queue . Bind ( <NUM_LIT:0> , queue , exchange , routing_key , <EOL> nowait , arguments or dict ( ) ) , <EOL> callback , replies ) <EOL> def queue_declare ( self , callback , <EOL> queue = '<STR_LIT>' , <EOL> passive = False , <EOL> durable = False , <EOL> exclusive = False , <EOL> auto_delete = False , <EOL> nowait = False , <EOL> arguments = None ) : <EOL> """<STR_LIT>""" <EOL> if queue : <EOL> condition = ( spec . Queue . DeclareOk , <EOL> { '<STR_LIT>' : queue } ) <EOL> else : <EOL> condition = spec . Queue . DeclareOk <EOL> replies = [ condition ] if nowait is False else [ ] <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Queue . Declare ( <NUM_LIT:0> , queue , passive , durable , <EOL> exclusive , auto_delete , nowait , <EOL> arguments or dict ( ) ) , <EOL> callback , replies ) <EOL> def queue_delete ( self , <EOL> callback = None , <EOL> queue = '<STR_LIT>' , <EOL> if_unused = False , <EOL> if_empty = False , <EOL> nowait = False ) : <EOL> """<STR_LIT>""" <EOL> replies = [ spec . Queue . DeleteOk ] if nowait is False else [ ] <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Queue . Delete ( <NUM_LIT:0> , queue , if_unused , if_empty , <EOL> nowait ) , <EOL> callback , replies ) <EOL> def queue_purge ( self , callback = None , queue = '<STR_LIT>' , nowait = False ) : <EOL> """<STR_LIT>""" <EOL> replies = [ spec . Queue . PurgeOk ] if nowait is False else [ ] <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Queue . Purge ( <NUM_LIT:0> , queue , nowait ) , callback , replies ) <EOL> def queue_unbind ( self , <EOL> callback = None , <EOL> queue = '<STR_LIT>' , <EOL> exchange = None , <EOL> routing_key = None , <EOL> arguments = None ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> if routing_key is None : <EOL> routing_key = queue <EOL> return self . _rpc ( spec . Queue . Unbind ( <NUM_LIT:0> , queue , exchange , routing_key , <EOL> arguments or dict ( ) ) , <EOL> callback , [ spec . Queue . UnbindOk ] ) <EOL> def tx_commit ( self , callback = None ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Tx . Commit ( ) , callback , [ spec . Tx . CommitOk ] ) <EOL> def tx_rollback ( self , callback = None ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Tx . Rollback ( ) , callback , [ spec . Tx . RollbackOk ] ) <EOL> def tx_select ( self , callback = None ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_channel_and_callback ( callback ) <EOL> return self . _rpc ( spec . Tx . Select ( ) , callback , [ spec . Tx . SelectOk ] ) <EOL> def _add_callbacks ( self ) : <EOL> """<STR_LIT>""" <EOL> self . callbacks . add ( self . channel_number , spec . Basic . GetEmpty , <EOL> self . _on_getempty , False ) <EOL> self . callbacks . add ( self . channel_number , spec . Basic . Cancel , <EOL> self . _on_cancel , False ) <EOL> self . callbacks . add ( self . channel_number , spec . Channel . Flow , <EOL> self . _on_flow , False ) <EOL> self . callbacks . add ( self . channel_number , spec . Channel . Close , <EOL> self . _on_close , True ) <EOL> def _add_on_cleanup_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . callbacks . add ( self . channel_number , self . _ON_CHANNEL_CLEANUP_CB_KEY , <EOL> callback , one_shot = True , only_caller = self ) <EOL> def _cleanup ( self ) : <EOL> """<STR_LIT>""" <EOL> self . callbacks . process ( self . channel_number , <EOL> self . _ON_CHANNEL_CLEANUP_CB_KEY , self , <EOL> self ) <EOL> self . _consumers = dict ( ) <EOL> self . callbacks . cleanup ( str ( self . channel_number ) ) <EOL> self . _cookie = None <EOL> def _cleanup_consumer_ref ( self , consumer_tag ) : <EOL> """<STR_LIT>""" <EOL> self . _consumers_with_noack . discard ( consumer_tag ) <EOL> self . _consumers . pop ( consumer_tag , None ) <EOL> self . _cancelled . discard ( consumer_tag ) <EOL> def _get_cookie ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _cookie <EOL> def _handle_content_frame ( self , frame_value ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> response = self . _content_assembler . process ( frame_value ) <EOL> except exceptions . UnexpectedFrameError : <EOL> self . _on_unexpected_frame ( frame_value ) <EOL> return <EOL> if response : <EOL> if isinstance ( response [ <NUM_LIT:0> ] . method , spec . Basic . Deliver ) : <EOL> self . _on_deliver ( * response ) <EOL> elif isinstance ( response [ <NUM_LIT:0> ] . method , spec . Basic . GetOk ) : <EOL> self . _on_getok ( * response ) <EOL> elif isinstance ( response [ <NUM_LIT:0> ] . method , spec . Basic . Return ) : <EOL> self . _on_return ( * response ) <EOL> def _on_cancel ( self , method_frame ) : <EOL> """<STR_LIT>""" <EOL> if method_frame . method . consumer_tag in self . _cancelled : <EOL> return <EOL> self . _cleanup_consumer_ref ( method_frame . method . consumer_tag ) <EOL> def _on_cancelok ( self , method_frame ) : <EOL> """<STR_LIT>""" <EOL> self . _cleanup_consumer_ref ( method_frame . method . consumer_tag ) <EOL> def _on_close ( self , method_frame ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . warning ( '<STR_LIT>' , <EOL> method_frame . method . reply_code , <EOL> method_frame . method . reply_text , <EOL> self ) <EOL> self . _send_method ( spec . Channel . CloseOk ( ) ) <EOL> if self . is_closing : <EOL> self . _closing_code_and_text = ( method_frame . method . reply_code , <EOL> method_frame . method . reply_text ) <EOL> else : <EOL> self . _set_state ( self . CLOSED ) <EOL> try : <EOL> self . callbacks . process ( self . channel_number , '<STR_LIT>' , <EOL> self , self , <EOL> method_frame . method . reply_code , <EOL> method_frame . method . reply_text ) <EOL> finally : <EOL> self . _cleanup ( ) <EOL> def _on_close_meta ( self , reply_code , reply_text ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' , self ) <EOL> if not self . is_closed : <EOL> self . _closing_code_and_text = reply_code , reply_text <EOL> self . _set_state ( self . CLOSED ) <EOL> try : <EOL> self . callbacks . process ( self . channel_number , '<STR_LIT>' , <EOL> self , self , <EOL> reply_code , <EOL> reply_text ) <EOL> finally : <EOL> self . _cleanup ( ) <EOL> def _on_closeok ( self , method_frame ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . info ( '<STR_LIT>' , method_frame . method , self ) <EOL> self . _set_state ( self . CLOSED ) <EOL> try : <EOL> self . callbacks . process ( self . channel_number , '<STR_LIT>' , <EOL> self , self , <EOL> self . _closing_code_and_text [ <NUM_LIT:0> ] , <EOL> self . _closing_code_and_text [ <NUM_LIT:1> ] ) <EOL> finally : <EOL> self . _cleanup ( ) <EOL> def _on_deliver ( self , method_frame , header_frame , body ) : <EOL> """<STR_LIT>""" <EOL> consumer_tag = method_frame . method . consumer_tag <EOL> if consumer_tag in self . _cancelled : <EOL> if self . is_open and consumer_tag not in self . _consumers_with_noack : <EOL> self . basic_reject ( method_frame . method . delivery_tag ) <EOL> return <EOL> if consumer_tag not in self . _consumers : <EOL> LOGGER . error ( '<STR_LIT>' , method_frame ) <EOL> return <EOL> self . _consumers [ consumer_tag ] ( self , method_frame . method , <EOL> header_frame . properties , body ) <EOL> def _on_eventok ( self , method_frame ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' , method_frame ) <EOL> def _on_flow ( self , _method_frame_unused ) : <EOL> """<STR_LIT>""" <EOL> if self . _has_on_flow_callback is False : <EOL> LOGGER . warning ( '<STR_LIT>' ) <EOL> def _on_flowok ( self , method_frame ) : <EOL> """<STR_LIT>""" <EOL> self . flow_active = method_frame . method . active <EOL> if self . _on_flowok_callback : <EOL> self . _on_flowok_callback ( method_frame . method . active ) <EOL> self . _on_flowok_callback = None <EOL> else : <EOL> LOGGER . warning ( '<STR_LIT>' ) <EOL> def _on_getempty ( self , method_frame ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' , method_frame ) <EOL> if self . _on_getok_callback is not None : <EOL> self . _on_getok_callback = None <EOL> def _on_getok ( self , method_frame , header_frame , body ) : <EOL> """<STR_LIT>""" <EOL> if self . _on_getok_callback is not None : <EOL> callback = self . _on_getok_callback <EOL> self . _on_getok_callback = None <EOL> callback ( self , method_frame . method , header_frame . properties , body ) <EOL> else : <EOL> LOGGER . error ( '<STR_LIT>' ) <EOL> def _on_openok ( self , method_frame ) : <EOL> """<STR_LIT>""" <EOL> if self . is_closing : <EOL> LOGGER . debug ( '<STR_LIT>' , method_frame ) <EOL> else : <EOL> self . _set_state ( self . OPEN ) <EOL> if self . _on_openok_callback is not None : <EOL> self . _on_openok_callback ( self ) <EOL> def _on_return ( self , method_frame , header_frame , body ) : <EOL> """<STR_LIT>""" <EOL> if not self . callbacks . process ( self . channel_number , '<STR_LIT>' , self , <EOL> self , <EOL> method_frame . method , <EOL> header_frame . properties , <EOL> body ) : <EOL> LOGGER . warning ( '<STR_LIT>' , <EOL> method_frame . method , header_frame . properties ) <EOL> def _on_selectok ( self , method_frame ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( "<STR_LIT>" , method_frame ) <EOL> def _on_synchronous_complete ( self , _method_frame_unused ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . debug ( '<STR_LIT>' , len ( self . _blocked ) ) <EOL> self . _blocking = None <EOL> while self . _blocked and self . _blocking is None : <EOL> self . _rpc ( * self . _blocked . popleft ( ) ) <EOL> def _rpc ( self , method , callback = None , acceptable_replies = None ) : <EOL> """<STR_LIT>""" <EOL> assert method . synchronous , ( <EOL> '<STR_LIT>' <EOL> % ( method , ) ) <EOL> if not isinstance ( acceptable_replies , ( type ( None ) , list ) ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> if callback is not None : <EOL> if not is_callable ( callback ) : <EOL> raise TypeError ( <EOL> '<STR_LIT>' ) <EOL> if not acceptable_replies : <EOL> raise ValueError ( <EOL> '<STR_LIT>' ) <EOL> if self . is_closed : <EOL> raise exceptions . ChannelClosed <EOL> if self . _blocking : <EOL> LOGGER . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> method , acceptable_replies ) <EOL> return self . _blocked . append ( [ method , callback , acceptable_replies ] ) <EOL> if acceptable_replies : <EOL> self . _blocking = method . NAME <EOL> LOGGER . debug ( <EOL> '<STR_LIT>' , <EOL> method , acceptable_replies ) <EOL> for reply in acceptable_replies : <EOL> if isinstance ( reply , tuple ) : <EOL> reply , arguments = reply <EOL> else : <EOL> arguments = None <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . callbacks . add ( self . channel_number , reply , <EOL> self . _on_synchronous_complete , <EOL> arguments = arguments ) <EOL> if callback is not None : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> self . callbacks . add ( self . channel_number , reply , callback , <EOL> arguments = arguments ) <EOL> self . _send_method ( method ) <EOL> def _send_method ( self , method , content = None ) : <EOL> """<STR_LIT>""" <EOL> self . connection . _send_method ( self . channel_number , method , content ) <EOL> def _set_cookie ( self , cookie ) : <EOL> """<STR_LIT>""" <EOL> self . _cookie = cookie <EOL> def _set_state ( self , connection_state ) : <EOL> """<STR_LIT>""" <EOL> self . _state = connection_state <EOL> def _on_unexpected_frame ( self , frame_value ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . error ( '<STR_LIT>' , frame_value ) <EOL> def _validate_channel_and_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_open : <EOL> raise exceptions . ChannelClosed ( ) <EOL> if callback is not None and not is_callable ( callback ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> class ContentFrameAssembler ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _method_frame = None <EOL> self . _header_frame = None <EOL> self . _seen_so_far = <NUM_LIT:0> <EOL> self . _body_fragments = list ( ) <EOL> def process ( self , frame_value ) : <EOL> """<STR_LIT>""" <EOL> if ( isinstance ( frame_value , frame . Method ) and <EOL> spec . has_content ( frame_value . method . INDEX ) ) : <EOL> self . _method_frame = frame_value <EOL> elif isinstance ( frame_value , frame . Header ) : <EOL> self . _header_frame = frame_value <EOL> if frame_value . body_size == <NUM_LIT:0> : <EOL> return self . _finish ( ) <EOL> elif isinstance ( frame_value , frame . Body ) : <EOL> return self . _handle_body_frame ( frame_value ) <EOL> else : <EOL> raise exceptions . UnexpectedFrameError ( frame_value ) <EOL> def _finish ( self ) : <EOL> """<STR_LIT>""" <EOL> content = ( self . _method_frame , self . _header_frame , <EOL> b'<STR_LIT>' . join ( self . _body_fragments ) ) <EOL> self . _reset ( ) <EOL> return content <EOL> def _handle_body_frame ( self , body_frame ) : <EOL> """<STR_LIT>""" <EOL> self . _seen_so_far += len ( body_frame . fragment ) <EOL> self . _body_fragments . append ( body_frame . fragment ) <EOL> if self . _seen_so_far == self . _header_frame . body_size : <EOL> return self . _finish ( ) <EOL> elif self . _seen_so_far > self . _header_frame . body_size : <EOL> raise exceptions . BodyTooLongError ( self . _seen_so_far , <EOL> self . _header_frame . body_size ) <EOL> return None <EOL> def _reset ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _method_frame = None <EOL> self . _header_frame = None <EOL> self . _seen_so_far = <NUM_LIT:0> <EOL> self . _body_fragments = list ( ) </s>
