)JfvdZddlZddlZddlmZddlmZmZm Z m Z ddl m Z ddl ZddlmZmZddlmZddlmZdd lmZejeZGd d ejZed gd ZGddeZGddejj Z!Gdde j"Z#Gddej$Z%dS)aUsing Pika with a Twisted reactor. The interfaces in this module are Deferred-based when possible. This means that the connection.channel() method and most of the channel methods return Deferreds instead of taking a callback argument and that basic_consume() returns a Twisted DeferredQueue where messages from the server will be stored. Refer to the docstrings for TwistedProtocolConnection.channel() and the TwistedChannel class for details. N) namedtuple)defererrorreactorprotocol)Failure) exceptionsspec)nbio_interface)check_callback_arg) ExchangeTypec6eZdZdZdfd ZdZdZdZxZS)ClosableDeferredQueuez Like the normal Twisted DeferredQueue, but after close() is called with an exception instance all pending Deferreds are errbacked and further attempts to call get() or put() return a Failure wrapping that exception. Nchd|_tt|||dSN)closedsuperr__init__)selfsizebacklog __class__s c/home/alex/cs2snipeproduction/venv/lib/python3.11/site-packages/pika/adapters/twisted_connection.pyrzClosableDeferredQueue.__init__(s1  #T**33D'BBBBBc|jr3tdtj|jStj||S)z} Like the original :meth:`DeferredQueue.put` method, but returns an errback if the queue is closed. z-Impossible to put to the queue, it is closed.)rLOGGERrrfail DeferredQueueput)robjs rrzClosableDeferredQueue.put,sL ; + LLH I I I:dk** *"&&tS111rc|jr3tdtj|jStj|S)a Returns a Deferred that will fire with the next item in the queue, when it's available. The Deferred will errback if the queue is closed. :returns: Deferred that fires with the next item. :rtype: Deferred z/Impossible to get from the queue, it is closed.)rrrrrrgetrs rr"zClosableDeferredQueue.get7sJ ; + LLJ K K K:dk** *"&&t,,,rc|jr td|j||_|jr3|j||j3g|_dS)zPCloses the queue. Errback the pending calls to :meth:`get()`. z)Queue was already closed with reason: %s.N)rrwarningwaitingpoperrbackpending)rreasons rclosezClosableDeferredQueue.closeGsv ; ( NNF; ( ( ( l / L     & &v . . .l / rNN) __name__ __module__ __qualname____doc__rrr"r+ __classcell__rs@rrr!sz CCCCCC 2 2 2---       rrReceivedMessagechannelmethod propertiesbodyceZdZdZdZdZdZdZdZdZ dZ d Z e d Z e d Ze d Ze d Ze dZe dZe dZdZdZd4dZd5dZ d6dZd7dZd8dZ d9dZd:dZd;dZd7d Zdd)Z(d*Z)d+Z*d?d,Z+ d@d-Z,dAd.Z-d/Z. dBd0Z/d1Z0d2Z1d3Z2dS)CTwistedChannela A wrapper around Pika's Channel. Channel methods that normally take a callback argument are wrapped to return a Deferred that fires with whatever would be passed to the callback. If the channel gets closed, all pending Deferreds are errbacked with a ChannelClosed exception. The returned Deferreds fire with whatever arguments the callback to the original method would receive. Some methods like basic_consume and basic_get are wrapped in a special way, see their docstrings for details. c||_d|_t|_i|_d|_|j|jtj j gdi|_ d|_ d|_ i|_d|_t!j|_|j|j|j|jdSNF)_channel_closedset_calls _consumers_basic_get_deferred add_callback _on_getemptyr BasicGetEmpty_queue_name_to_consumer_tags_delivery_confirmation_delivery_message_id _deliveries_puback_returnrDeferred on_closedadd_on_close_callback_on_channel_closedadd_on_cancel_callback _on_consumer_cancelled_by_broker)rr5s rrzTwistedChannel.__init__gs  ee #'  ""4#4tz7J6K#( * * *-/)&+#$(!#)) ++D,CDDD ,,  1 3 3 3 3 3rcNd|jj|jS)Nz<{cls} channel={chan!r}>)clschan)formatrr-r=r#s r__repr__zTwistedChannel.__repr__s+)00'dm1== =rc||_|jD]}||j|jD]}||j|jD]}||jt|_i|_i|_|j |jdSr) r>r@r(rJvaluesrAr+r?rMcallback)rr=r*dconsumers rrOz!TwistedChannel._on_channel_closeds  $ $A IIdl # # # #!((** $ $A IIdl # # # #..00 ) )H NN4< ( ( ( (ee   -----rc,||S)awCalled by impl when broker cancels consumer via Basic.Cancel. This is a RabbitMQ-specific feature. The circumstances include deletion of queue being consumed as well as failure of a HA node responsible for the queue being consumed. :param pika.frame.Method method_frame: method frame with the `spec.Basic.Cancel` method )_on_consumer_cancelled)r method_frames rrQz/TwistedChannel._on_consumer_cancelled_by_brokers**<888rc`|jj}||jvrtd||S|j|t j|j|=|j D](} | |#t$rY%wxYw|S)aCalled when the broker cancels a consumer via Basic.Cancel or when the broker responds to a Basic.Cancel request by Basic.CancelOk. :param pika.frame.Method frame: method frame with the `spec.Basic.Cancel` or `spec.Basic.CancelOk` method z%basic_cancel - consumer not found: %s) r6 consumer_tagrArr%r+r ConsumerCancelledrGrXremoveKeyError)rframer`ctagss rr]z%TwistedChannel._on_consumer_cancelleds|0 t . . NNB' ) ) )L  %++J,H,J,JKKK OL )6==??  E  \****     sB B+*B+c|jtddS|jddS)z3Callback the Basic.Get deferred with None. Nz7Got Basic.GetEmpty but no Basic.Get calls were pending.)rBrr%rY)r _method_frames rrDzTwistedChannel._on_getemptysJ  # + NN+ , , , F  ))$/////rcttj|tjfd}|S)a#Wrap Pika's Channel method to make it return a Deferred that fires when the method completes and errbacks if the channel gets closed. If the original method's callback would receive more than one argument, the Deferred fires with a tuple of argument values. cHjrtjjStjjjfd}||d< |i|n##t$rtjcYSwxYwS)Nct|dkr$t|dSj|dS)z Make sure that the deferred is called with a single argument. In case the original callback fires with more than one, convert to a tuple. N)lenrYtuple)argsrZs rsingle_argumentzMTwistedChannel._wrap_channel_method..wrapped..single_argumentsG t99q==JJuT{{+++++AJ%%%%rrY) r>rrrLr@add addCallback _clear_call Exception)rnkwargsrorZr6rs @rwrappedz4TwistedChannel._wrap_channel_method..wrappeds| 0z$,///  A KOOA    MM$*A . . . & & & & &"1F:  $''''' $ $ $z||### $Hs6A??BB)getattrr= functoolswraps)rnamerur6s` @r_wrap_channel_methodz#TwistedChannel._wrap_channel_methodsQ--        !  6rc:|j||Srr@discardrretrZs rrrzTwistedChannel._clear_call A rc|jjSr)r=channel_numberr#s rrzTwistedChannel.channel_numbers }++rc|jjSr)r= connectionr#s rrzTwistedChannel.connections }''rc|jjS)zFReturns True if the channel is closed. :rtype: bool )r= is_closedr#s rrzTwistedChannel.is_closeds}&&rc|jjS)zoReturns True if client-initiated closing of the channel is in progress. :rtype: bool )r= is_closingr#s rrzTwistedChannel.is_closings}''rc|jjS)zDReturns True if the channel is open. :rtype: bool )r=is_openr#s rrzTwistedChannel.is_open s}$$rc|jjSr)r= flow_activer#s rrzTwistedChannel.flow_actives }((rc|jjSr)r= consumer_tagsr#s rrzTwistedChannel.consumer_tagss }**rcF|j|j|dS)a Pass in a Deferred and a list replies from the RabbitMQ broker which you'd like the Deferred to be callbacked on with the frame as callback value. :param Deferred deferred: The Deferred to callback :param list replies: The replies to callback on N)r=rCrY)rdeferredrepliess rcallback_deferredz TwistedChannel.callback_deferreds% ""8#4g>>>>>rcFjfddS)aPass a callback function that will be called when a published message is rejected and returned by the server via `Basic.Return`. :param callable callback: The method to call on callback with the message as only argument. The message is a named tuple with the following attributes - channel: this TwistedChannel - method: pika.spec.Basic.Return - properties: pika.spec.BasicProperties - body: bytes c<t|||SNr4r3)r=r6r7r8rYrs rz7TwistedChannel.add_on_return_callback..9s1xx !) 88rN)r=add_on_return_callbackrrYs``rrz%TwistedChannel.add_on_return_callback,sC ,,      rrFc:|j||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. ) delivery_tagmultiple)r= basic_ack)rrrs rrzTwistedChannel.basic_ackCs($}&&%':: :rcx|d}||}||jS)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. This method wraps :meth:`Channel.basic_cancel ` and closes any deferred queue associated with that consumer. :param str consumer_tag: Identifier for the consumer :returns: Deferred that fires on the Basic.CancelOk response :rtype: Deferred :raises ValueError: basic_cancel)r`)rzrqr])rr`rurZs rrzTwistedChannel.basic_cancelXs>*++N;; G . . .}}T8999rNc l jrtjjSt tjj fd} fd} j||||||n##t$rtjcYSwxYwS)alConsume from a server queue. Sends the AMQP 0-9-1 command Basic.Consume to the broker and binds messages for the consumer_tag to a :class:`ClosableDeferredQueue`. If you do not pass in a consumer_tag, one will be automatically generated for you. 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 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 :returns: Deferred that fires with a tuple ``(queue_object, consumer_tag)``. The Deferred will errback with an instance of :class:`exceptions.ChannelClosed` if the call fails. The queue object is an instance of :class:`ClosableDeferredQueue`, where data received from the queue will be stored. Clients should use its :meth:`get() ` method to fetch an individual message, which will return a Deferred firing with a namedtuple whose attributes are: - channel: this TwistedChannel - method: pika.spec.Basic.Deliver - properties: pika.spec.BasicProperties - body: bytes :rtype: Deferred c |jj}jt |j|<j |fdSr) r6r`rG setdefaultr?rprAr@r}rY)rdr`rZqueue queue_objrs r on_consume_okz3TwistedChannel.basic_consume..on_consume_oks{ <4L  - 8 8suu  !c,///,5DOL ) K   " " " JJ <0 1 1 1 1 1rcTt|||dS)zjAdd the ReceivedMessage to the queue, while replacing the channel implementation. r4N)rr3)r=r6r7r8rrs ron_message_callbackz9TwistedChannel.basic_consume..on_message_callbacksC MM !)      r)rrauto_ack exclusiver` argumentsrY) r>rrrrLr@rpr= basic_consumers) rrrrr`rrrrZrs `` @@rrzTwistedChannel.basic_consumeqs T < ,:dl++ +)++ N    2 2 2 2 2 2 2 2       M ' '$7!#)#& (       :<<    s/!BB10B1cjtjfd}fd}d||}|||||_|S)aGet a single message from the AMQP broker. Will return If the queue is empty, it will return None. 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 This method wraps :meth:`Channel.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 bool auto_ack: Tell the broker to not expect a reply :returns: Deferred that fires with a namedtuple whose attributes are: - channel: this TwistedChannel - method: pika.spec.Basic.GetOk - properties: pika.spec.BasicProperties - body: bytes If the queue is empty, None will be returned. :rtype: Deferred :raises pika.exceptions.DuplicateGetOkCallback: Nc@|dS|\}}}}t|||Srr)resultr=r6r7r8rs rcreate_namedtuplez3TwistedChannel.basic_get..create_namedtuples>~t17 .Hfj$"%  rcd_|Sr)rBrrs rcleanup_attributez3TwistedChannel.basic_get..cleanup_attributes'+D $Mr basic_get)rr)rBr DuplicateGetOkCallbackrzrqaddBoth)rrrrrrZs` rrzTwistedChannel.basic_gets<  # /355 5           3D % %k 2 2( , , , '((( #$$$#$ rTc<|j|||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. )rrrequeue)r= basic_nack)rrrrs rrzTwistedChannel.basic_nacks,&}''%(   rc:|jrtj|jS|j|||||}|jstj|S|xjdz c_tj|j |j<|j |jS)aPublish to the channel with the given exchange, routing key and body. This method wraps :meth:`Channel.basic_publish `, but makes sure the channel is not closed before publishing. 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 :returns: A Deferred that fires with the result of the channel's basic_publish. :rtype: Deferred :raises UnroutableError: raised when a message published in publisher-acknowledgments mode (see `BlockingChannel.confirm_delivery`) is returned via `Basic.Return` followed by `Basic.Ack`. :raises NackError: raised when a message published in publisher-acknowledgements mode is Nack'ed by the broker. See `BlockingChannel.confirm_delivery`. )exchange routing_keyr8r7 mandatoryrk) r>rrr= basic_publishrHsucceedrIrLrJ)rrrr8r7rrs rrzTwistedChannel.basic_publishsB < ,:dl++ +,,#! -!! * ?=(( (  % % * % %:?.:J:JD T6 7#D$=> >rcD|d|||S)aSpecify quality of service. This method requests a specific quality of service. The QoS can be specified for the current channel or for all channels on the connection. 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. 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 apply to all channels on the connection. :returns: Deferred that fires on the Basic.QosOk response :rtype: Deferred basic_qos) prefetch_sizeprefetch_count global_qosrz)rrrrs rrzTwistedChannel.basic_qosFs5@6t((55')!    rc:|j||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 )rr)r= basic_reject)rrrs rrzTwistedChannel.basic_rejectls(}))%w*88 8rc@|d|S)a^This 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. :returns: Deferred that fires on the Basic.RecoverOk response :rtype: Deferred basic_recover)rr)rrs rrzTwistedChannel.basic_recover|s%:t((99'JJJJrNormal shutdownc:|j||S)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 ) reply_code reply_text)r=r+rrrs rr+zTwistedChannel.closes}""jZ"PPPrc0jr.tdtjdSd}|j}fd}||j j |S)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: http://www.rabbitmq.com/confirms.html#publisher-confirms :returns: Deferred that fires on the Confirm.SelectOk response :rtype: Deferred z3confirm_delivery: confirmation was already enabled.Nconfirm_delivery)ack_nack_callbackcXd_d_td|S)NTrzDelivery confirmation enabled.)rHrIrdebugrs rset_delivery_confirmationzBTwistedChannel.confirm_delivery..set_delivery_confirmations+*.D '()D % LL9 : : :Mr) rHrrrrrz_on_delivery_confirmationrqr=r_on_puback_message_returned)rrurZrs` rrzTwistedChannel.confirm_deliverys  & ' LLN O O O=&& &++,>?? Gd&D E E E      /000 ,,T-MNNNrc:|jj|jvrtddS|jjr(fd|jD}|ng}|D],}|j|}|j|=t|jtj j j rgt d|j|j |j|jg}d|_ng}|t!j|t|jtj j jsJ|j8|jg}d|_|t!j|||j.dS)aInvoked by pika when RabbitMQ responds to a Basic.Publish RPC command, passing in either a Basic.Ack or Basic.Nack frame with the delivery tag of the message that was published. The delivery tag is an integer counter indicating the message number that was sent on the channel via Basic.Publish. Here we're just doing house keeping to keep track of stats and remove message numbers that we expect a delivery confirmation of from the list used to keep track of messages that are pending confirmation. :param pika.frame.Method method_frame: Basic.Ack or Basic.Nack frame z3Delivery tag %s not found in the pending deliveriesNc g|] }|k| Sr).0tagrs r z.s#KKKCsl7J7JC7J7J7Jrz3Message was Nack'ed by broker: nack=%r; channel=%s;)r6rrJrrrsort isinstancepikar rENackr%rrKr(r NackErrorAckUnroutableErrorrY)rr^tagsrrZreturned_messagesrs @rrz(TwistedChannel._on_delivery_confirmations$*7 t/ / / LLN% ' ' ' F   ' "KKKK4#3KKKD IIKKKK >D 4 4C %A %,-ty/CDD 4I ')<>>>&2)-)<(=%*.D''(*% *./@AABBBB!,"5ty7JKKKKK&2)-)<(=%*.D'IIj89JKKLLLLJJ|23333/ 4 4rc Ft|tjjs J|t|tjs J|t d|j|j|||t|nd|t|||||_ dS)aCalled as the result of Basic.Return from broker in publisher-acknowledgements mode. Saves the info as a ReturnedMessage instance in self._puback_return. :param pika.Channel channel: our self._impl channel :param pika.spec.Basic.Return method: :param pika.spec.BasicProperties properties: message properties :param bytes body: returned message body; empty string if no body zPublished message was returned: _delivery_confirmation=%s; channel=%s; method=%r; properties=%r; body_size=%d; body_prefix=%.255rNr4) rr rEReturnBasicPropertiesrr%rHrrlr3rK)rr5r6r7r8s rrz*TwistedChannel._on_puback_message_returneds&$*"344<`, and removes the reference to the queue object after it gets deleted on the server. :param str queue: The queue to delete :param bool if_unused: only delete if it's unused :param bool if_empty: only delete if the queue is empty :returns: Deferred that fires on the Queue.DeleteOk response :rtype: Deferred :raises ValueError: queue_delete)rrif_emptyc0tj|tD]_}j|t jd|zj|=j||`|S)NzQueue %s was deleted.) listrGr"r?rAr+r rarb)r queue_namer`rs r_clear_consumerz4TwistedChannel.queue_delete.._clear_consumers $599*ceeLL!N!N " "  -330/*<>>???OL11*=DD """"Jr)rzrq)rrrrrurZr s` rrzTwistedChannel.queue_deletes] ++N;; G%9x H H H     }}_e444rc@|d|S)zPurge all of the messages from the specified queue :param str queue: The queue to purge :returns: Deferred that fires on the Queue.PurgeOk response :rtype: Deferred :raises ValueError: queue_purge)rr)rrs rrzTwistedChannel.queue_purges%8t((77eDDDDrcF|d||||S)aUnbind a queue from an exchange. :param str queue: The queue to unbind from the exchange :param str exchange: The source exchange to bind from :param str routing_key: The routing key to unbind :param dict arguments: Custom key/value pair arguments for the binding :returns: Deferred that fires on the Queue.UnbindOk response :rtype: Deferred :raises ValueError: queue_unbindrrrs rrzTwistedChannel.queue_unbinds7 9t((88#     rc<|dS)zCommit a transaction. :returns: Deferred that fires on the Tx.CommitOk response :rtype: Deferred :raises ValueError: tx_commitrr#s rrzTwistedChannel.tx_commits 6t((55777rc<|dS)zRollback a transaction. :returns: Deferred that fires on the Tx.RollbackOk response :rtype: Deferred :raises ValueError: tx_rollbackrr#s rrzTwistedChannel.tx_rollbacks 8t((77999rc<|dS)aVSelect standard transaction mode. This method sets the channel to use standard transactions. The client must use this method at least once on a channel before using the Commit or Rollback methods. :returns: Deferred that fires on the Tx.SelectOk response :rtype: Deferred :raises ValueError: tx_selectrr#s rrzTwistedChannel.tx_selects 6t((55777r)rF)r)FFNN)F)NFTr<)rrF)T)rr)rN)NNrNr,)FFFFN)FF)NNN)3r-r.r/r0rrVrOrQr]rDrzrrpropertyrrrrrrrrrrrrrrrrrrr+rrrrr directrrrrrrrrrrrrrrrrr:r:Zs  3332===...$ 9 9 90000%%%N ,,X,((X(''X'((X(%%X%))X)++X+ ? ? ?   .::::*::::6 % %#' $ RRRRh5555n    :"& % /?/?/?/?b$ $ $ $ L8888 K K K K Q Q Q Q<.4.4.4`AAA0>@ $    ,(4':!&!&%*"'#'( ( ( ( T     %)#$&"&     .@@@ $$$    *$# %"' $ " " " " H5555@ E E E#!%#     .888::: 8 8 8 8 8rr:cXeZdZdZfdZdZdZdZdZdZ dZ d Z d Z d Z xZS) _TwistedConnectionAdaptera8A Twisted-specific implementation of a Pika Connection. NOTE: since `base_connection.BaseConnection`'s primary responsibility is management of the transport, we use `pika.connection.Connection` directly as our base class because this adapter uses a different transport management strategy. ctt|||||d|pt|_d|_dS)NF) parameterson_open_callbackon_open_error_callbackon_close_callbackinternal_connection_workflow)rrrr_reactor _transport)rrrrrcustom_reactorrs rrz"_TwistedConnectionAdapter.__init__sS '..77!-#9/). 8 0 0 0'1' rcrt|dt|j||S)zXImplement :py:meth:`pika.connection.Connection._adapter_call_later()`. rY)r _TimerHandler! callLater)rdelayrYs r_adapter_call_laterz-_TwistedConnectionAdapter._adapter_call_later's4 8Z000DM33E8DDEEErc.|dS)z\Implement :py:meth:`pika.connection.Connection._adapter_remove_timeout()`. N)cancel)r timeout_ids r_adapter_remove_timeoutz1_TwistedConnectionAdapter._adapter_remove_timeout/s rcZt|d|j|dS)zeImplement :py:meth:`pika.connection.Connection._adapter_add_callback_threadsafe()`. rYN)r r!callFromThreadrs r _adapter_add_callback_threadsafez:_TwistedConnectionAdapter._adapter_add_callback_threadsafe6s0 8Z000 $$X.....rct)zImplement pure virtual :py:ref:meth:`pika.connection.Connection._adapter_connect_stream()` method. NOTE: This should not be called due to our initialization of Connection via `internal_connection_workflow=False` )NotImplementedErrorr#s r_adapter_connect_streamz1_TwistedConnectionAdapter._adapter_connect_stream>s "!rc8|jdS)zImplement pure virtual :py:ref:meth:`pika.connection.Connection._adapter_disconnect_stream()` method. N)r"loseConnectionr#s r_adapter_disconnect_streamz4_TwistedConnectionAdapter._adapter_disconnect_streamHs &&(((((rc:|j|dS)zoImplement pure virtual :py:ref:meth:`pika.connection.Connection._adapter_emit_data()` method. N)r"writerdatas r_adapter_emit_dataz,_TwistedConnectionAdapter._adapter_emit_dataPs d#####rc<||_|dS)zIntroduces transport to protocol after transport is connected. :param twisted.internet.interfaces.ITransport transport: :raises Exception: Exception-based exception on error N)r"_on_stream_connectedr transports rconnection_madez)_TwistedConnectionAdapter.connection_madeWs#$ !!#####rcd|_|j}t|tjr ||_d}t | tj n tj d|| |dS)a Called upon loss or closing of TCP connection. NOTE: `connection_made()` and `connection_lost()` are each called just once and in that order. All other callbacks are called between them. :param Failure: A Twisted Failure instance wrapping an exception. Nzconnection_lost: %r) r"valuer twisted_errorConnectionDone_errorrlogloggingDEBUGERROR_on_stream_terminated)rrs rconnection_lostz)_TwistedConnectionAdapter.connection_lostbsv  e]9 : : DKE EM7==w}(% 1 1 1 ""5)))))rc0||dS)zCalled to deliver incoming data from the server to the protocol. :param data: Non-empty data bytes. :raises Exception: Exception-based exception on error N)_on_data_availabler8s r data_receivedz'_TwistedConnectionAdapter.data_receivedts %%%%%r)r-r.r/r0rr(r,r/r2r5r:r?rJrMr1r2s@rrrs     FFF///""")))$$$ $ $ $***$&&&&&&&rrceZdZdZddZddZedZedZdd Z d Z e j fd Z d Zd ZdZddZdZdZdS)TwistedProtocolConnectionawA Pika-specific implementation of a Twisted Protocol. Allows using Twisted's non-blocking connectTCP/connectSSL methods for connecting to the server. TwistedProtocolConnection objects have a `ready` instance variable that's a Deferred which fires when the connection is ready to be used (the initial AMQP handshaking has been done). You *have* to wait for this Deferred to fire before requesting a channel. Once the connection is ready, you will be able to use the `closed` instance variable: a Deferred which fires when the connection is closed. Since it's Twisted handling connection establishing it does not accept connect callbacks, you have to implement that within Twisted. Also remember that the host, port and ssl values of the connection parameters are ignored because, yet again, it's Twisted who manages the connection. Nctj_jfdd_t |jjj|_ t_ dS)Nc,Sr)connectionReady)_rs rrz4TwistedProtocolConnection.__init__..s)=)=)?)?r)rrrrr#) rrLreadyrqrr_on_connection_ready_on_connection_failed_on_connection_closed_implr?r@)rrr#s` rrz"TwistedProtocolConnection.__init__sx^%%  ????@@@ .!!6#'#="8)     ee rctj}|j||j|j|||j||tS)aCreate a new channel with the next available channel number or pass in a channel number to use. Must be non-zero if you would like to specify but it is recommended that you let Pika manage the channel numbers. :param int channel_number: The channel number to use, defaults to the next available. :returns: a Deferred that fires with an instance of a wrapper around the Pika Channel class. :rtype: Deferred ) rrLrXr5rYr@rprqrrr:)rrrZs rr5z!TwistedProtocolConnection.channelsh N   >1:666  d&***}}^,,,rc|jjSr)rXrr#s rrz!TwistedProtocolConnection.is_opensz!!rc|jjSr)rXrr#s rrz#TwistedProtocolConnection.is_closedsz##rrc^|jjs|j|||jSr)rXrr+rrs rr+zTwistedProtocolConnection.closes/z# 5 J  Z 4 4 4{rc:|j|dSr)rXrMr8s r dataReceivedz&TwistedProtocolConnection.dataReceiveds   &&&&&rc|j||jdc}|_|r||dSdSr)rXrJrTr()rr*rZs rconnectionLostz(TwistedProtocolConnection.connectionLostsQ ""6*** D 4:   IIf       rcz|j|tj||dSr)rXr?rProtocolmakeConnectionr=s rrdz(TwistedProtocolConnection.makeConnections7 ""9---((y99999rc|S)zLThis method will be called when the underlying connection is ready. rr#s rrRz)TwistedProtocolConnection.connectionReadys  rc|jdc}|_|r/tj|_|ddSdSr)rTrrLrrY)r _connectionrZs rrUz.TwistedProtocolConnection._on_connection_readysJ D 4:  .**DK JJt       rc|jdc}|_|r<|jjj}t j|}||dSdSr)rTrXparamsconnection_attemptsr AMQPConnectionErrorr()rrg_error_messagerZattemptsexcs rrVz/TwistedProtocolConnection._on_connection_failedsS D 4:  z(rs  """"""OOOOOOOOOOOO******!!!!!!!!......DDDDDD++++++  8 $ $22222E/222j*.HHHJJt8t8t8t8t8Vt8t8t8nj&j&j&j&j& :j&j&j&Zxxxxx 1xxxv     >8     r