<s> """<STR_LIT>""" <EOL> import logging <EOL> import math <EOL> from pamqp import body as pamqp_body <EOL> from pamqp import header as pamqp_header <EOL> from pamqp import specification as pamqp_spec <EOL> from amqpstorm import compatibility <EOL> from amqpstorm . base import FRAME_MAX <EOL> from amqpstorm . base import Handler <EOL> from amqpstorm . exception import AMQPChannelError <EOL> from amqpstorm . exception import AMQPInvalidArgument <EOL> from amqpstorm . exception import AMQPMessageError <EOL> from amqpstorm . message import Message <EOL> LOGGER = logging . getLogger ( __name__ ) <EOL> class Basic ( Handler ) : <EOL> """<STR_LIT>""" <EOL> def qos ( self , prefetch_count = <NUM_LIT:0> , prefetch_size = <NUM_LIT:0> , global_ = False ) : <EOL> """<STR_LIT>""" <EOL> if not compatibility . is_integer ( prefetch_count ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not compatibility . is_integer ( prefetch_size ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not isinstance ( global_ , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> qos_frame = pamqp_spec . Basic . Qos ( prefetch_count = prefetch_count , <EOL> prefetch_size = prefetch_size , <EOL> global_ = global_ ) <EOL> return self . _channel . rpc_request ( qos_frame ) <EOL> def get ( self , queue = '<STR_LIT>' , no_ack = False , to_dict = True ) : <EOL> """<STR_LIT>""" <EOL> if not compatibility . is_string ( queue ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not isinstance ( no_ack , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif self . _channel . consumer_tags : <EOL> raise AMQPChannelError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> get_frame = pamqp_spec . Basic . Get ( queue = queue , <EOL> no_ack = no_ack ) <EOL> with self . _channel . lock and self . _channel . rpc . lock : <EOL> message = self . _get_message ( get_frame ) <EOL> if message and to_dict : <EOL> return message . to_dict ( ) <EOL> return message <EOL> def recover ( self , requeue = False ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( requeue , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> recover_frame = pamqp_spec . Basic . Recover ( requeue = requeue ) <EOL> return self . _channel . rpc_request ( recover_frame ) <EOL> def consume ( self , callback = None , queue = '<STR_LIT>' , consumer_tag = '<STR_LIT>' , <EOL> exclusive = False , no_ack = False , no_local = False , arguments = None ) : <EOL> """<STR_LIT>""" <EOL> if not compatibility . is_string ( queue ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not compatibility . is_string ( consumer_tag ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not isinstance ( exclusive , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not isinstance ( no_ack , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not isinstance ( no_local , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif arguments is not None and not isinstance ( arguments , dict ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> self . _channel . consumer_callback = callback <EOL> consume_frame = pamqp_spec . Basic . Consume ( queue = queue , <EOL> consumer_tag = consumer_tag , <EOL> exclusive = exclusive , <EOL> no_local = no_local , <EOL> no_ack = no_ack , <EOL> arguments = arguments ) <EOL> result = self . _channel . rpc_request ( consume_frame ) <EOL> consumer_tag = result [ '<STR_LIT>' ] <EOL> self . _channel . add_consumer_tag ( consumer_tag ) <EOL> return consumer_tag <EOL> def cancel ( self , consumer_tag = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if not compatibility . is_string ( consumer_tag ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> cancel_frame = pamqp_spec . Basic . Cancel ( consumer_tag = consumer_tag ) <EOL> result = self . _channel . rpc_request ( cancel_frame ) <EOL> self . _channel . remove_consumer_tag ( consumer_tag ) <EOL> return result <EOL> def publish ( self , body , routing_key , exchange = '<STR_LIT>' , properties = None , <EOL> mandatory = False , immediate = False ) : <EOL> """<STR_LIT>""" <EOL> self . _validate_publish_parameters ( body , exchange , immediate , mandatory , <EOL> properties , routing_key ) <EOL> properties = properties or { } <EOL> body = self . _handle_utf8_payload ( body , properties ) <EOL> properties = pamqp_spec . Basic . Properties ( ** properties ) <EOL> method_frame = pamqp_spec . Basic . Publish ( exchange = exchange , <EOL> routing_key = routing_key , <EOL> mandatory = mandatory , <EOL> immediate = immediate ) <EOL> header_frame = pamqp_header . ContentHeader ( body_size = len ( body ) , <EOL> properties = properties ) <EOL> send_buffer = [ method_frame , header_frame ] <EOL> for body_frame in self . _create_content_body ( body ) : <EOL> send_buffer . append ( body_frame ) <EOL> if self . _channel . confirming_deliveries : <EOL> with self . _channel . rpc . lock : <EOL> return self . _publish_confirm ( send_buffer ) <EOL> self . _channel . write_frames ( send_buffer ) <EOL> def ack ( self , delivery_tag = None , multiple = False ) : <EOL> """<STR_LIT>""" <EOL> if delivery_tag is not None and not compatibility . is_integer ( delivery_tag ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> elif not isinstance ( multiple , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> ack_frame = pamqp_spec . Basic . Ack ( delivery_tag = delivery_tag , <EOL> multiple = multiple ) <EOL> self . _channel . write_frame ( ack_frame ) <EOL> def nack ( self , delivery_tag = None , multiple = False , requeue = True ) : <EOL> """<STR_LIT>""" <EOL> if delivery_tag is not None and not compatibility . is_integer ( delivery_tag ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> elif not isinstance ( multiple , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not isinstance ( requeue , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> nack_frame = pamqp_spec . Basic . Nack ( delivery_tag = delivery_tag , <EOL> multiple = multiple , <EOL> requeue = requeue ) <EOL> self . _channel . write_frame ( nack_frame ) <EOL> def reject ( self , delivery_tag = None , requeue = True ) : <EOL> """<STR_LIT>""" <EOL> if delivery_tag is not None and not compatibility . is_integer ( delivery_tag ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> elif not isinstance ( requeue , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> reject_frame = pamqp_spec . Basic . Reject ( delivery_tag = delivery_tag , <EOL> requeue = requeue ) <EOL> self . _channel . write_frame ( reject_frame ) <EOL> @ staticmethod <EOL> def _validate_publish_parameters ( body , exchange , immediate , mandatory , <EOL> properties , routing_key ) : <EOL> """<STR_LIT>""" <EOL> if not compatibility . is_string ( body ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not compatibility . is_string ( routing_key ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not compatibility . is_string ( exchange ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif properties and not isinstance ( properties , dict ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not isinstance ( mandatory , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> elif not isinstance ( immediate , bool ) : <EOL> raise AMQPInvalidArgument ( '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def _handle_utf8_payload ( body , properties ) : <EOL> """<STR_LIT>""" <EOL> if compatibility . is_unicode ( body ) : <EOL> if '<STR_LIT>' not in properties : <EOL> properties [ '<STR_LIT>' ] = '<STR_LIT:utf-8>' <EOL> encoding = properties [ '<STR_LIT>' ] <EOL> body = body . encode ( encoding ) <EOL> elif compatibility . PYTHON3 and isinstance ( body , str ) : <EOL> body = bytes ( body , encoding = '<STR_LIT:utf-8>' ) <EOL> return body <EOL> def _get_message ( self , get_frame ) : <EOL> """<STR_LIT>""" <EOL> uuid_get = self . _channel . rpc . register_request ( get_frame . valid_responses ) <EOL> uuid_header = self . _channel . rpc . register_request ( [ '<STR_LIT>' ] ) <EOL> uuid_body = self . _channel . rpc . register_request ( [ '<STR_LIT>' ] ) <EOL> self . _channel . write_frame ( get_frame ) <EOL> get_frame = self . _channel . rpc . get_request ( uuid_get , raw = True ) <EOL> if not isinstance ( get_frame , pamqp_spec . Basic . GetOk ) : <EOL> self . _channel . rpc . remove ( uuid_header ) <EOL> self . _channel . rpc . remove ( uuid_body ) <EOL> return None <EOL> content_header = self . _channel . rpc . get_request ( uuid_header , raw = True ) <EOL> body = self . _get_content_body ( uuid_body , content_header . body_size ) <EOL> return Message ( channel = self . _channel , <EOL> body = body , <EOL> method = dict ( get_frame ) , <EOL> properties = dict ( content_header . properties ) ) <EOL> def _publish_confirm ( self , send_buffer ) : <EOL> """<STR_LIT>""" <EOL> confirm_uuid = self . _channel . rpc . register_request ( [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> self . _channel . write_frames ( send_buffer ) <EOL> result = self . _channel . rpc . get_request ( confirm_uuid , raw = True ) <EOL> self . _channel . check_for_errors ( ) <EOL> if isinstance ( result , pamqp_spec . Basic . Ack ) : <EOL> return True <EOL> elif isinstance ( result , pamqp_spec . Basic . Nack ) : <EOL> return False <EOL> else : <EOL> raise AMQPMessageError ( '<STR_LIT>' <EOL> % ( result , dict ( result ) ) ) <EOL> @ staticmethod <EOL> def _create_content_body ( body ) : <EOL> """<STR_LIT>""" <EOL> frames = int ( math . ceil ( len ( body ) / float ( FRAME_MAX ) ) ) <EOL> for offset in compatibility . RANGE ( <NUM_LIT:0> , frames ) : <EOL> start_frame = FRAME_MAX * offset <EOL> end_frame = start_frame + FRAME_MAX <EOL> if end_frame > len ( body ) : <EOL> end_frame = len ( body ) <EOL> yield pamqp_body . ContentBody ( body [ start_frame : end_frame ] ) <EOL> def _get_content_body ( self , uuid_body , body_size ) : <EOL> """<STR_LIT>""" <EOL> body = bytes ( ) <EOL> while len ( body ) < body_size : <EOL> body_piece = self . _channel . rpc . get_request ( uuid_body , raw = True , <EOL> multiple = True ) <EOL> if not body_piece : <EOL> break <EOL> body += body_piece . value <EOL> self . _channel . rpc . remove ( uuid_body ) <EOL> return body </s>
