)JfdZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z ddlmZmZmZddlmZejeZdZGddeZGd d eZdS) zThe Channel class provides a wrapper for interacting with RabbitMQ implementing the methods and behaviors for an AMQP Channel. N)Enum) unicode_typedictkeys is_integer) ExchangeTypeiceZdZdZdZdZdZdZedededed iZd Z d Z d Z d Z dTdZ dZdZdZdZdUdZdVdZ dWdZdZdXdZdYdZ dZdZ d[dZd\dZd]d Zd^d"Zd_d#Zed$Z d`d%Z e!j"ddddddfd&Z#dad'Z$ dbd(Z%d_d)Z&ed*Z'ed+Z(ed,Z)ed-Z*d.Z+ dcd/Z, ddd0Z- ded1Z.d_d2Z/ dfd3Z0d_d4Z1d_d5Z2d_d6Z3d7Z4d8Z5d9Z6d:Z7d;Z8d<Z9d=Z:d>Z;d?ZdBZ?dCZ@dDZAdEZBdFZCdGZDdHZEdIZFdJZGdKZHdLZIdMZJdgdNZKdOZLd_dPZMdQZNdRZOdSZPdS)hChannelaA Channel is the primary communication method for interacting with RabbitMQ. It is recommended that you do not directly invoke the creation of a channel object in your application code but rather construct a channel by calling the active connection's channel() method. rCLOSEDOPENINGOPENCLOSING_on_channel_cleanupct|tstj|t j|||_|j|_||_d|_ t|_ tj t|_d|_d|_t%|_t)|_t%|_d|_d|_||_|j|_d|_d|_dS)asCreate a new instance of the Channel :param pika.connection.Connection connection: The connection :param int channel_number: The channel number for this instance :param callable on_open_callback: The callback to call on channel open. The callback will be invoked with the `Channel` instance as its only argument. TNF) isinstanceint exceptionsInvalidChannelNumber validatorsrpc_completion_callbackchannel_number callbacks connection flow_activeContentFrameAssembler_content_assembler collectionsdequelist_blocked _blocking_has_on_flow_callbackset _cancelleddict _consumers_consumers_with_noack_on_flowok_callback_on_getok_callback_on_openok_callbackr _state_closing_reason_cookie)selfrron_open_callbacks O/home/alex/cs2snipeproduction/venv/lib/python3.11/site-packages/pika/channel.py__init__zChannel.__init__2s.#.. B1.AA A*+;<<<,#-$ "7"9"9#)$&&11 %*"%%&&%(UU"#' "&#3 k  $ c|jS)zNReturn the channel object as its channel number :rtype: int )rr0s r2__int__zChannel.__int__`s ""r4c hd|jjd|jd|j|jd|jd S)N) __class____name__r _STATE_NAMESr-rr6s r2__repr__zChannel.__repr__hsB N # # #T%8%8%8  dk * * *DOOO= =r4TcT|D]$}|j|j|||%dS)aPass in a callback handler and a list replies from the RabbitMQ broker which you'd like the callback notified of. Callbacks should allow for the frame parameter to be passed in. :param callable callback: The callback to call :param list replies: The replies to get a callback for :param bool one_shot: Only handle the first type callback Nraddr)r0callbackrepliesone_shotreplys r2 add_callbackzChannel.add_callbackmsC O OE N  t2E8X N N N N O Or4ch|j|jtjj|ddS)a Pass a callback function that will be called when the basic_cancel is sent by the server. The callback function should receive a frame parameter. :param callable callback: The callback to call on Basic.Cancel from broker FN)rrBrspecBasicCancelr0rCs r2add_on_cancel_callbackzChannel.add_on_cancel_callbackzs9 4. 0A8  " " " " "r4cL|j|jd|d|dS)aPass a callback function that will be called when the channel is closed. The callback function will receive the channel and an exception describing why the channel was closed. If the channel is closed by broker via Channel.Close, the callback will receive `ChannelClosedByBroker` as the reason. If graceful user-initiated channel closing completes successfully ( either directly of indirectly by closing a connection containing the channel) and closing concludes gracefully without Channel.Close from the broker and without loss of connection, the callback will receive `ChannelClosedByClient` exception as reason. If channel was closed due to loss of connection, the callback will receive another exception type describing the failure. :param callable callback: The callback, having the signature: callback(Channel, Exception reason) _on_channel_closeFNrArLs r2add_on_close_callbackzChannel.add_on_close_callbacks7* 4.0CX $ ( ( ( ( (r4cvd|_|j|jtjj|ddS)a Pass a callback function that will be called when Channel.Flow is called by the remote server. Note that newer versions of RabbitMQ will not issue this but instead use TCP backpressure :param callable callback: The callback function TFN)r$rrBrrIr FlowrLs r2add_on_flow_callbackzChannel.add_on_flow_callbacksA&*" 4. 0A8  " " " " "r4cJ|j|jd|ddS)aMPass a callback function that will be called when basic_publish is sent a message that has been rejected and returned by the server. :param callable callback: The function to call, having the signature callback(channel, method, properties, body) where - channel: pika.channel.Channel - method: pika.spec.Basic.Return - properties: pika.spec.BasicProperties - body: bytes _on_returnFNrArLs r2add_on_return_callbackzChannel.add_on_return_callbacks) 4. hNNNNNr4Fc||tj||S)aAcknowledge one or more messages. When sent by the client, this method acknowledges one or more messages delivered via the Deliver or Get-Ok methods. When sent by server, this method acknowledges one or more messages published with the Publish method on a channel in confirm mode. The acknowledgement can be for a single message or a set of messages up to and including a specific message. :param integer delivery_tag: int/long The server-assigned delivery tag :param bool multiple: If set to True, the delivery tag is treated as "up to and including", so that multiple messages can be acknowledged with a single method. If set to False, the delivery tag refers to a single message. If the multiple field is 1, and the delivery tag is zero, this indicates acknowledgement of all outstanding messages. )_raise_if_not_open _send_methodrIrJAck)r0 delivery_tagmultiples r2 basic_ackzChannel.basic_acks:$ !!!   h!G!GHHHr4Nctj|d|tj|}||jvrt d|dS||jvrt d|dSt d|||r|j|=|0|j |j tj j||j ||tj |||s|jnd|stj jd|ifgngdS)aThis method cancels a consumer. This does not affect already delivered messages, but it does mean the server will not send any more messages for that consumer. The client may receive an arbitrary number of messages in between sending the cancel method and receiving the cancel-ok reply. It may also be sent from the server to the client in the event of the consumer being unexpectedly cancelled (i.e. cancelled for any reason other than the server receiving the corresponding basic.cancel from the client). This allows clients to be notified of the loss of consumers due to events such as queue deletion. :param str consumer_tag: Identifier for the consumer :param callable callback: callback(pika.frame.Method) for method Basic.CancelOk. If None, do not expect a Basic.CancelOk response, otherwise, callback must be callable :raises ValueError: consumer_tagz1basic_cancel - consumer is already cancelling: %sNz%basic_cancel - consumer not found: %sz#Cancelling consumer: %s (nowait=%s))r`nowait)rrequire_stringrXrr&LOGGERwarningr(debugrrBrrIrJCancelOk_rpcrK _on_cancelok)r0r`rCras r2 basic_cancelzChannel.basic_cancels& !,??? !!!3H== 4? * * NNN' ) ) ) F t . . NNB' ) ) ) F :L     .  -   N  t2DJ4G' ) ) ) L))) $*##f#MM+1;$##t$,DJ'$l*)+  - - - - -r4c 8tj|dtj||tj||s|}||jvs ||jvrtj ||r|j |||j|<||j n|}| tj|||||p t#|tjjd|ifg|S)aSends the AMQP 0-9-1 command Basic.Consume to the broker and binds messages for the consumer_tag to the consumer callback. If you do not pass in a consumer_tag, one will be automatically generated for you. Returns the consumer tag. For more information on basic_consume, see: Tutorial 2 at http://www.rabbitmq.com/getstarted.html http://www.rabbitmq.com/confirms.html http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.consume :param str queue: The queue to consume from. Use the empty string to specify the most recent server-named queue for this channel :param callable on_message_callback: The function to call when consuming with the signature on_message_callback(channel, method, properties, body), where - channel: pika.channel.Channel - method: pika.spec.Basic.Deliver - properties: pika.spec.BasicProperties - body: bytes :param bool auto_ack: if set to True, automatic acknowledgement mode will be used (see http://www.rabbitmq.com/confirms.html). This corresponds with the 'no_ack' parameter in the basic.consume AMQP 0.9.1 method :param bool exclusive: Don't allow other consumers on the queue :param str consumer_tag: Specify your own consumer tag :param dict arguments: Custom key/value pair arguments for the consumer :param callable callback: callback(pika.frame.Method) for method Basic.ConsumeOk. :returns: Consumer tag which may be used to cancel the consumer. :rtype: str :raises ValueError: queueN)rkr`no_ack exclusive argumentsr`)rrbrequire_callbackrXr_generate_consumer_tagr(r&rDuplicateConsumerTagr)rB _on_eventokrgrIrJConsumer' ConsumeOk) r0rkon_message_callbackauto_ackrmr`rnrC rpc_callbacks r2 basic_consumezChannel.basic_consumesAR !%111#$7888 !!!*8444 96688L 4? * *ldo.M.M1,?? ?  9  & * *< 8 8 8(; %+3+;t''  J  U,8&.)2)2) >@Lj" %     r4cFd|jtjjfzS)zGenerate a consumer tag NOTE: this protected method may be called by derived classes :returns: consumer tag :rtype: str z ctag%i.%s)ruuiduuid4hexr6s r2rpzChannel._generate_consumer_tagPs d14:<<3CDDDr4ctj|dtj||jt j||_|tj ||dS)a6Get a single message from the AMQP broker. If you want to be notified of Basic.GetEmpty, use the Channel.add_callback method adding your Basic.GetEmpty callback which should expect only one parameter, frame. Due to implementation details, this cannot be called a second time until the callback is executed. For more information on basic_get and its parameters, see: http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.get :param str queue: The queue from which to get a message. Use the empty string to specify the most recent server-named queue for this channel :param callable callback: The callback to call with a message that has the signature callback(channel, method, properties, body), where: - channel: pika.channel.Channel - method: pika.spec.Basic.GetOk - properties: pika.spec.BasicProperties - body: bytes :param bool auto_ack: Tell the broker to not expect a reply :raises ValueError: rkN)rkrl) rrbror+rDuplicateGetOkCallbackrYrIrJGet)r0rkrCrvs r2 basic_getzChannel.basic_get[sv. !%111#H---  " .355 5"* $*..uX.FFGGGGGr4c||tj|||S)aLThis method allows a client to reject one or more incoming messages. It can be used to interrupt and cancel large incoming messages, or return untreatable messages to their original queue. :param integer delivery_tag: int/long The server-assigned delivery tag :param bool multiple: If set to True, the delivery tag is treated as "up to and including", so that multiple messages can be acknowledged with a single method. If set to False, the delivery tag refers to a single message. If the multiple field is 1, and the delivery tag is zero, this indicates acknowledgement of all outstanding messages. :param bool requeue: If requeue is true, the server will attempt to requeue the message. If requeue is false or the requeue attempt fails the messages are discarded or dead-lettered. )rXrYrIrJNack)r0r[r\requeues r2 basic_nackzChannel.basic_nack~sC& !!!  JOOL(G < <>> >r4c|t|tr|d}|pt j}|tj|||||fdS)aPublish to the channel with the given exchange, routing key and body. For more information on basic_publish and what the parameters do, see: http://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.publish :param str exchange: The exchange to publish to :param str routing_key: The routing key to bind on :param bytes body: The message body :param pika.spec.BasicProperties properties: Basic.properties :param bool mandatory: The mandatory flag zutf-8)exchange routing_key mandatoryN) rXrrencoderIBasicPropertiesrYrJPublish)r0rrbody propertiesrs r2 basic_publishzChannel.basic_publishs$ !!! dL ) ) (;;w''D94#7#9#9   J  +6)2  4 46@$5G I I I I Ir4c2|tj|tjd|tjd||t j||||t jjgS)aSpecify quality of service. This method requests a specific quality of service. The client can request that messages be sent in advance so that when the client finishes processing a message, the following message is already held locally, rather than needing to be sent down the channel. The QoS can be applied separately to each new consumer on channel or shared across all consumers on the channel. Prefetching gives a performance improvement. :param int prefetch_size: This field specifies the prefetch window size. The server will send a message in advance if it is equal to or smaller in size than the available prefetch size (and also falls into other prefetch limits). May be set to zero, meaning "no specific limit", although other prefetch limits may still apply. The prefetch-size is ignored by consumers who have enabled the no-ack option. :param int prefetch_count: Specifies a prefetch window in terms of whole messages. This field may be used in combination with the prefetch-size field; a message will only be sent in advance if both prefetch windows (and those at the channel and connection level) allow it. The prefetch-count is ignored by consumers who have enabled the no-ack option. :param bool global_qos: Should the QoS be shared across all consumers on the channel. :param callable callback: The callback to call for Basic.QosOk response :raises ValueError: prefetch_sizeprefetch_count) rXrrzero_or_greaterrgrIrJQosQosOk)r0rr global_qosrCs r2 basic_qoszChannel.basic_qossH !!!*8444"?MBBB"#3^DDDyy JNN=.* E Ex Z     r4c|t|std|tj||S)a_Reject an incoming message. This method allows a client to reject a message. It can be used to interrupt and cancel large incoming messages, or return untreatable messages to their original queue. :param integer delivery_tag: int/long The server-assigned delivery tag :param bool requeue: If requeue is true, the server will attempt to requeue the message. If requeue is false or the requeue attempt fails the messages are discarded or dead-lettered. :raises: TypeError zdelivery_tag must be an integer)rXr TypeErrorrYrIrJReject)r0r[rs r2 basic_rejectzChannel.basic_rejects[ !!!,'' ?=>> >  !2!2<!I!IJJJr4c|tj||tj||tjjgS)aThis method asks the server to redeliver all unacknowledged messages on a specified channel. Zero or more messages may be redelivered. This method replaces the asynchronous Recover. :param bool requeue: If False, the message will be redelivered to the original recipient. If True, the server will attempt to requeue the message, potentially then delivering it to an alternative subscriber. :param callable callback: Callback to call when receiving Basic.RecoverOk :param callable callback: callback(pika.frame.Method) for method Basic.RecoverOk :raises ValueError: )rXrrrgrIrJRecover RecoverOk)r0rrCs r2 basic_recoverzChannel.basic_recovers\ !!!*8444yy++G44h*./11 1r4Normal shutdownc|js|jr|td|||t j|||_t|j D]!}||j vr| |"| |j |tj||dd|jtjjgdS)aInvoke a graceful shutdown of the channel with the AMQP Broker. If channel is OPENING, transition to CLOSING and suppress the incoming Channel.OpenOk, if any. :param int reply_code: The reason code to send to broker :param str reply_text: The reason text to send to broker :raises ChannelWrongStateError: if channel is closed or closing zClosing channel (%s): %r on %s)r`rN) is_closed is_closingrXrcinforChannelClosedByClientr.rr(r&ri _set_staterrgrIr Close _on_closeokCloseOk)r0 reply_code reply_textr`s r2closez Channel.closes > &T_ &  # # % % %  4j*    *?   $ $%T_55 = =L4?22!!|!<<<  %%% $,$$ZQBB"T\%9$: < < < < t|std|tj|}|jjr |jjstj d|j |j tjj|d|j |j tjj|d|tj|||stjjgngdS)aTurn on Confirm mode in the channel. Pass in a callback to be notified by the Broker when a message has been confirmed as received or rejected (Basic.Ack, Basic.Nack) from the broker to the publisher. For more information see: https://www.rabbitmq.com/confirms.html :param callable ack_nack_callback: Required callback for delivery confirmations that has the following signature: callback(pika.frame.Method), where method_frame contains either method `spec.Basic.Ack` or `spec.Basic.Nack`. :param callable callback: callback(pika.frame.Method) for method Confirm.SelectOk :raises ValueError: zSconfirm_delivery requires a callback to receieve Basic.Ack/Basic.Nack notificationsz&Confirm.Select not Supported by ServerFN)callable ValueErrorrXrrrpublisher_confirmsrrMethodNotImplementedrrBrrIrJrZrrgConfirmSelectSelectOk)r0ack_nack_callbackrCras r2confirm_deliveryzChannel.confirm_delivery+s/")** ONOO O !!!3H==2 :* :18:: : 4. ,e 5 5 5 4. ,e 5 5 5 $,%%f--x17?4<())R A A A A Ar4c*t|jS)zaProperty method that returns a list of currently active consumers :rtype: list )rr(r6s r2 consumer_tagszChannel.consumer_tagsSs(((r4c \|tj|dtj|dtj|}|t jd|||||p t||st jj gngS)aBind an exchange to another exchange. :param str destination: The destination exchange to bind :param str source: The source exchange to bind to :param str routing_key: The routing key to bind on :param dict arguments: Custom key/value pair arguments for the binding :param callable callback: callback(pika.frame.Method) for method Exchange.BindOk :raises ValueError: destinationsourcer) rXrrbrrgrIExchangeBindr'BindOkr0rrrrnrCras r2 exchange_bindzChannel.exchange_bind\s !!!!+}===!&(3333H==yy M  q+v{F(2DFF 4 45=*0 8T] ! " "b:: :r4c ptj|d|tj|} t |t r|j}|tj d||||||| |p t || stj j gngS)aThis method creates an exchange if it does not already exist, and if the exchange exists, verifies that it is of the correct and expected class. If passive set, the server will reply with Declare-Ok if the exchange already exists with the same name, and raise an error if not and if the exchange does not already exist, the server MUST raise a channel exception with reply code 404 (not found). :param str exchange: The exchange name consists of a non-empty sequence of these characters: letters, digits, hyphen, underscore, period, or colon :param str exchange_type: The exchange type to use :param bool passive: Perform a declare or just check to see if it exists :param bool durable: Survive a reboot of RabbitMQ :param bool auto_delete: Remove when no more queues are bound to it :param bool internal: Can only be published to by other exchanges :param dict arguments: Custom key/value pair arguments for the exchange :param callable callback: callback(pika.frame.Method) for method Exchange.DeclareOk :raises ValueError: rr) rrbrXrrrvaluergrIrDeclarer' DeclareOk) r0r exchange_typepassivedurable auto_deleteinternalrnrCras r2exchange_declarezChannel.exchange_declareus> !(J777 !!!3H== mT * * 0)/Myy M ! !!X}gw"-xB)"&&& * *+3-3 ;T] $ % % == =r4c|tj|}|tjd|||||stjjgngS)a Delete the exchange. :param str exchange: The exchange name :param bool if_unused: only delete if the exchange is unused :param callable callback: callback(pika.frame.Method) for method Exchange.DeleteOk :raises ValueError: r)rXrrrgrIrDeleteDeleteOk)r0r if_unusedrCras r2exchange_deletezChannel.exchange_deletest !!!3H==yy--a9.4667?9?G$-011RII Ir4c |tj|}|tjd|||||||stjjgngS)aUnbind an exchange from another exchange. :param str destination: The destination exchange to unbind :param str source: The source exchange to unbind from :param str routing_key: The routing key to unbind :param dict arguments: Custom key/value pair arguments for the binding :param callable callback: callback(pika.frame.Method) for method Exchange.UnbindOk :raises ValueError: r)rXrrrgrIrUnbindUnbindOkrs r2exchange_unbindzChannel.exchange_unbindsv !!!3H==yy M Kf!* , ,-5,2 :T] # $ $<< d L L L&& s4#677888 r4c|j||j|d|j|dS)zRemove any references to the consumer tag in internal structures for consumer state. :param str consumer_tag: The consumer tag to cleanup N)r)discardr(popr&)r0r`s r2_cleanup_consumer_refzChannel._cleanup_consumer_refsO "**<888 L$///  -----r4c|jS)zUsed by the wrapper implementation (e.g., `BlockingChannel`) to retrieve the cookie that it set via `_set_cookie` :returns: opaque cookie value that was set via `_set_cookie` :rtype: object r/r6s r2 _get_cookiezChannel._get_cookies |r4c |j|}n+#tj$r||YdSwxYw|rt |djtjj r |j |dSt |djtjj r |j |dSt |djtjj r|j|dSdSdS)aThis is invoked by the connection when frames that are not registered with the CallbackManager have been found. This should only be the case when the frames are related to content delivery. The _content_assembler will be invoked which will return the fully formed message in three parts when all of the body frames have been received. :param pika.amqp_object.Frame frame_value: The frame to deliver Nr)rr rUnexpectedFrameError_on_unexpected_framermethodrIrJDeliver _on_deliverGetOk _on_getokReturnrU)r0 frame_valueresponses r2_handle_content_framezChannel._handle_content_frames .66{CCHH.     % %k 2 2 2 FF   +(1+,dj.@AA +  (++++HQK. 0@AA +))))HQK. 0ABB +****  + +  + +s$AAcn|jj|jvrdS||jjdS)zWhen the broker cancels a consumer, delete it from our internal dictionary. :param pika.frame.Method method_frame: The method frame received N)rr`r&rr0 method_frames r2rzChannel._on_cancels<   +t > > F ""<#6#CDDDDDr4cD||jjdS)zCalled in response to a frame from the Broker when the client sends Basic.Cancel :param pika.frame.Method method_frame: The method frame received N)rrr`r!s r2rhzChannel._on_canceloks$ ""<#6#CDDDDDr4c |jrJ|jJ||j |j|jd|||j|dS#|wxYw)a1Common logic for transitioning the channel to the CLOSED state: Set state to CLOSED, dispatch callbacks registered via `Channel.add_on_close_callback()`, and mop up. Assumes that the channel is not in CLOSED state and that `self._closing_reason` has been set up NrO)rr.rr rr rr r6s r2_transition_to_closedzChannel._transition_to_closeds>!!!#///  $$$  N " "4#68K#'t/C E E E MMOOOOODMMOOOOs (A,,Bctd|jj|jj||jrJ|tj tj |jj|jj|_ |j r|dS|dS)zHandle `Channel.Close` from broker. :param pika.frame.Method method_frame: Method frame with Channel.Close method z,Received remote Channel.Close (%s): %r on %sN)rcrdrrrrrYrIr rrChannelClosedByBrokerr.r&_drain_blocked_methods_on_remote_closer%r!s r2rzChannel._on_close_from_broker1s E#*5#*5t = = =>!!! $,..00111 *?   *L,?,J L L ? )  7 7 9 9 9 9 9  & & ( ( ( ( (r4ctd|||js||_|dSdS)aHandle meta-close request from either a remote Channel.Close from the broker (when a pending Channel.Close method is queued for execution) or a Connection's cleanup logic after sudden connection loss. We use this opportunity to transition to CLOSED state, clean up the channel, and dispatch the on-channel-closed callbacks. :param Exception reason: Exception describing the reason for closing. zHandling meta-close on %s: %rN)rcrerr.r%)r0reasons r2_on_close_metazChannel._on_close_metaVsO  4dFCCC~ )#)D  & & ( ( ( ( ( ) )r4cptd|j||dS)zInvoked when RabbitMQ replies to a Channel.Close method :param pika.frame.Method method_frame: Method frame with Channel.CloseOk method zReceived %s on %sN)rcrrr%r!s r2rzChannel._on_closeokfs6  ') ? ? ? ? ? / .r4c|jj|_|jr(||jjd|_dStddS)zCalled in response to us asking the server to toggle on Channel.Flow :param pika.frame.Method method_frame: The method frame received Nz0Channel.FlowOk received with no active callbacks)rrrr*rcrdr!s r2rzChannel._on_flowoks_ (.5  # O  $ $\%8%? @ @ @'+D $ $ $ NNM N N N N Nr4c\td||j d|_dSdS)zWhen we receive an empty reply do nothing but log it :param pika.frame.Method method_frame: The method frame received zReceived Basic.GetEmpty: %rN)rcrer+r!s r2rzChannel._on_getemptys8  2LAAA  " .&*D # # # / .r4c|j(|j}d|_|||j|j|dStddS)aCalled in reply to a Basic.Get when there is a message. :param pika.frame.Method method_frame: The method frame received :param pika.frame.Header header_frame: The header frame received :param bytes body: The body received Nz,Basic.GetOk received with no active callback)r+rrrcr.)r0r"r/rrCs r2rzChannel._on_getoksX  " ..H&*D # HT<. 0G N N N N N LLG H H H H Hr4c|jrtd|dS||j|j||dSdS)aCalled by our callback handler when we receive a Channel.OpenOk and subsequently calls our _on_openok_callback which was passed into the Channel constructor. The reason we do this is because we want to make sure that the on_open_callback parameter passed into the Channel constructor is not the first callback we make. Suppress the state transition and callback if channel is already in CLOSING state. :param pika.frame.Method method_frame: Channel.OpenOk frame z&Suppressing while in closing state: %sN)rrcrerrr,r!s r2rzChannel._on_openoksh ? / LLA< P P P P P OODI & & &'3((.....43r4c |j|jd|||j|j|s(t d|j|jdSdS)zCalled if the server sends a Basic.Return frame. :param pika.frame.Method method_frame: The Basic.Return frame :param pika.frame.Header header_frame: The content header frame :param bytes body: The message body rUz*Basic.Return received from server (%r, %r)N)rr rrrrcre)r0r"r/rs r2rUzChannel._on_returns{~%%d&9<&*L,?&2&=tEE H LLE&-|/F H H H H H H Hr4c<td|dS)zCalled when the broker sends a Confirm.SelectOk frame :param pika.frame.Method method_frame: The method frame received zConfirm.SelectOk Received: %rNr1r!s r2 _on_selectokzChannel._on_selectoks  4lCCCCCr4ctdt|jd|_|jr8|j5|j|j|jr |j/dSdSdSdS)aThis is called when a synchronous command is completed. It will undo the blocking state and send all the frames that stacked up while we were in the blocking state. :param pika.frame.Method method_frame_unused: The method frame received z%i blocked framesN)rcrelenr"r#rgpopleftr3s r2_on_synchronous_completez Channel._on_synchronous_completes  (#dm*<*<===m 0 6 DIt},,.. / /m 0 6 6 6 0 0 0 0 6 6r4chtdt|j|jr}|jd}t |t jjr| |j ntd||j{dSdS)aThis is called when the broker sends a Channel.Close while the client is in CLOSING state. This method checks the blocked method queue for a pending client-initiated Channel.Close method and ensures its callbacks are processed, but does not send the method to the broker. The broker may close the channel before responding to outstanding in-transit synchronous methods, or even before these methods have been sent to the broker. AMQP 0.9.1 obliges the server to drop all methods arriving on a closed channel other than Channel.CloseOk and Channel.Close. Since the response to a synchronous method that blocked the channel never arrives, the channel never becomes unblocked, and the Channel.Close, if any, in the blocked queue has no opportunity to be sent, and thus its completion callback would never be called. z@Draining %i blocked frames due to broker-requested Channel.Closerz#Ignoring drained blocked method: %sN) rcrer=r"r>rrIr rr+r.)r0rs r2r(z.Channel._drain_blocked_methods_on_remote_closes"  N      m L]**,,Q/F&$,"455 L##D$89999 BFKKK m L L L L Lr4cx|js Jd|t|tdtfst d|/t |st d|st d|jr||j r;t d|||j |||gdS|||r|j|_ t d|||D]}t|t r|\}}nd}t d|j|j||j| |=t d |j|j||| dSdS) aMake a synchronous channel RPC call for a synchronous method frame. If the channel is already in the blocking state, then enqueue the request, but don't send it at this time; it will be eventually sent by `_on_synchronous_complete` after the prior blocking request receives a response. If the channel is not in the blocking state and `acceptable_replies` is not empty, transition the channel to the blocking state and register for `_on_synchronous_complete` before sending the request. NOTE: A callback must be accompanied by non-empty acceptable_replies. :param pika.amqp_object.Method method: The AMQP method to invoke :param callable callback: The callback for the RPC response :param list|None acceptable_replies: A (possibly empty) sequence of replies this RPC call expects or None z8Only synchronous-capable methods may be used with _rpc: Nz)acceptable_replies should be list or Nonez%callback should be None or a callablez8Unexpected callback for asynchronous (nowait) operation.zIAlready in blocking state, so enqueueing method %s; acceptable_replies=%rz:Entering blocking state on frame %s; acceptable_replies=%rz'Adding on_synchronous_complete callback)rnz&Adding passed-in RPC response callback) synchronousrtyper!rrrrrXr#rcrer"appendrYNAMEtuplerrBrr?)r0rrCacceptable_repliesrFrns r2rgz Channel._rpcs:$!    V     ,tDzz4.@AA IGHH H  H%% I GHHH& P NPPP > &  # # % % % >  LL()/1C E E E M &(4F!G H H H F &!!!  <#[DN LLL* , , ,, < <eU++%',$E99 $I FGGG""4#6#(#'#@-6#888'LL!IJJJN&&t':','/1:'<<<' < < < rr3r7r?rGrMrPrSrVr]rirxrprrrrrrrrpropertyrrrdirectrrrrrrrrrrrrrrrrrrrr rrrrrhr%rr+rrrrrrrrrrUr;r?r(rgrXrYrNrrr4r2r r sFG DG  f L"7,,,\###=== O O O O " " "(((0 " " " O O OIIII*8-8-8-8-z % %#' $#FFFFP E E E!H!H!H!HF>>>>6"& % IIII8!"!"" * * * * XKKKK$1111*&<&<&<&Create a new instance of the conent frame assembler. Nr _method_frame _header_frame _seen_so_farr!_body_fragmentsr6s r2r3zContentFrameAssembler.__init__s."!#vvr4ct|tjr'tj|jjr ||_dSt|tjr(||_ |j dkr| SdSt|tj r| |Stj|)a!Invoked by the Channel object when passed frames that are not setup in the rpc process and that don't have explicit reply types defined. This includes Basic.Publish, Basic.GetOk and Basic.Return :param Method|Header|Body frame_value: The frame to process Nr)rframeMethodrI has_contentrINDEXr\Headerr] body_size_finishBody_handle_body_framerrrRs r2r zContentFrameAssembler.processs {EL 1 1 ? !3!9:: ?!,D 4  U\ 2 2 ?!,D $))||~~%t  UZ 0 0 ?**;77 71+>> >r4c||j|jd|jf}||S)z}Invoked when all of the message has been received :rtype: tuple(pika.frame.Method, pika.frame.Header, str) r4)r\r]joinr__reset)r0rKs r2rgzContentFrameAssembler._finishs: %t'988D0113 r4cL|xjt|jz c_|j|j|j|jjkr|S|j|jjkr$tj |j|jjdS)aReceive body frames and append them to the stack. When the body size matches, call the finish method. :param Body body_frame: The body frame :raises: pika.exceptions.BodyTooLongError :rtype: tuple(pika.frame.Method, pika.frame.Header, str)|None N) r^r=fragmentr_rDr]rfrgrBodyTooLongError)r0 body_frames r2riz(ContentFrameAssembler._handle_body_frames S!4555 ##J$7888   2 < < <<<>> !  !3!= = =-d.?.2.@.JLL Ltr4cVd|_d|_d|_t|_dS)z&Reset the values for processing framesNrr[r6s r2rlzContentFrameAssembler._resets,!!#vvr4N) r=rSrTrUr3r rgrirlrXr4r2rrsi &&&???.   $&&&&&r4r)rUrloggingrzenumr pika.framerapika.exceptionsr pika.specrIpika.validatorsr pika.compatrrrpika.exchange_typer getLoggerr=rc MAX_CHANNELSobjectr rrXr4r2r}sA $$$$$$$$$$$$::::::::::++++++  8 $ $ G:G:G:G:G:fG:G:G:T,H&H&H&H&H&FH&H&H&H&H&r4