o wvXj#6 @sdZddlmZmZddlZddlZddlZddlmZddl m Z ddl m Z ddl mZddlmZmZmZmZdd lmZmZd d lmZmZd d lmZmZd d lmZd dlm Z!d dl"mZzddl#m$Z$Wn!e%yzddl&m$Z$Wne%yddl'm$Z$YnwYnwdZ(dZ)ee*Z+da,ddZ-ddZ.d=ddZ/GdddeZ0ddZ1d>dd Z2d!d"Z3d#d$Z4d%d&Z5d?d'd(Z6  d?d)d*Z7d@d+d,Z8 dAd-d.Z9d/d0Z:d1d2Z;e d3d4ZdCd9d:Z?Gd;d<dZs    zBroadcast.__init__)NNFTNN)__name__ __module__ __qualname____doc__rattrsr> __classcell__r"r"rBr#rEs rcCs||jjjvSr) connectionclientdeclared_entities)entitychannelr"r"r#declaration_cachedqsrOFcKs |r t||fi|St||S)zDeclare entity (cached).)_imaybe_declare_maybe_declare)rMrNretry retry_policyr"r"r#rus rcCs0|j}|s|std||||}|SdS)zMake sure the channel is bound to the entity. :param entity: generic kombu nomenclature, generally an exchange or queue :param channel: channel to bind to the entity :return: the updated entity z#Cannot bind channel {} to entity {}N)is_boundr r<bind)rMrNrTr"r"r#_ensure_channel_is_bound|s  rVcCs|}t|||dur|jstd||j}d}}|jr1|jr1|jjj}t |}||vr1dS|js8t d|j |d|durI|rI| ||durQ|j |_ dS)Nz(channel is None and entity {} not bound.Fchannel disconnected)rNT)rVrTr r<rNrJcan_cache_declarationrKrLhashr declareaddr9)rMrNorigdeclaredidentr"r"r#rQs,      rQcKs:t|||jjs td|jjjj|tfi|||S)NrW)rVrNrJr rKensurerQ)rMrNrSr"r"r#rPs  rPc #stfdd}|g|pg|_|'t|jjj||ddD]}zVWq ty2Yq wWddS1s>wYdS)z&Drain messages from consumer instance.cs||fdSr)append)bodymessageaccr"r# on_messagesz"drain_consumer..on_messageT)limittimeoutignore_timeoutsN)r callbacksrrNrJrKpopleft IndexError)consumerrfrgrire_r"rcr#rs    "rcKs$t|jd|g|d||||dS)zIterator over messages.)queuesrN)rfrgriNr")rConsumer)connrNr4rfrgrirAr"r"r#rsrc csN|rt|p tD]}z |j|dVWq tjy$|r"|s"Yq wdS)aBest practice generator wrapper around ``Connection.drain_events``. Able to drain events forever, with a limit, and optionally ignoring timeout errors (a timeout of 1 is often used in environments where the socket can get "stuck", and is a best practice for Kombu consumers). ``eventloop`` is a generator. Examples: >>> from kombu.common import eventloop >>> def run(conn): ... it = eventloop(conn, timeout=1, ignore_timeouts=True) ... next(it) # one event consumed, or timed out. ... ... for _ in eventloop(conn, timeout=1, ignore_timeouts=True): ... pass # loop forever. It also takes an optional limit parameter, and timeout errors are propagated by default:: for _ in eventloop(connection, limit=1, timeout=1): pass See Also: :func:`itermessages`, which is an event loop bound to one or more consumers, that yields any messages received. )rgN)rr drain_eventssocketrg)rprfrgrhir"r"r#rsrc KsH|j|f|||dt|jd|jdtj|j|jdfi|S)aSend reply for request. Arguments: exchange (kombu.Exchange, str): Reply exchange req (~kombu.Message): Original request, a message with a ``reply_to`` property. producer (kombu.Producer): Producer instance retry (bool): If true must retry according to the ``reply_policy`` argument. retry_policy (Dict): Retry settings. **props (Any): Extra properties. )r;rRrSreply_tocorrelation_id) routing_keyru serializercontent_encoding)publishdict propertiesget serializers type_to_name content_typerx)r;reqmsgproducerrRrSpropsr"r"r#rs   rc os||dd}d}z*t|||g|Ri|D]\}}|s!|d}|VqW|r2||jdSdS|r=||jww)z,Generator collecting replies from ``queue``.no_ackTFN) setdefaultrackafter_reply_message_receivedr9) rprNr4argsrArreceivedrarbr"r"r#r s&  rcCstjd||dddS)Nz#Connection error: %r. Retry in %ss T)exc_info)loggererror)excintervalr"r"r#_ensure_errbacks rc cs,zdVWdS|j|jyYdSwr)connection_errorschannel_errors)rpr"r"r#_ignore_errors"s  rcOsB|rt|||i|WdS1swYt|S)aIgnore connection and channel errors. The first argument must be a connection object, or any other object with ``connection_error`` and ``channel_error`` attributes. Can be used as a function: .. code-block:: python def example(connection): ignore_errors(connection, consumer.channel.close) or as a context manager: .. code-block:: python def example(connection): with ignore_errors(connection): consumer.channel.close() Note: Connection and channel errors should be properly handled, and not ignored. Using this function is only acceptable in a cleanup phase, like when a connection is lost or at shutdown. N)r)rpfunrrAr"r"r# ignore_errors*s    rcCs|r||dSdSrr")rJrN on_reviver"r"r#revive_connectionKs rc Ks|pt}|jdd4}|j|d|j}tt||d} |j||f|| d|} | |it||d\} } | WdS1sAwYdS)zFunction wrapper to handle connection errors. Ensures function performing broker commands completes despite intermittent connection failures. T)block)errback)r)rr)rJN)racquireensure_connectiondefault_channelrr autoretryrz) poolrrrArroptsrprNreviverretvalrmr"r"r#rPs  $rc@s@eZdZdZdZddZdddZddd Zd d Zd d Z dS)QoSaThread safe increment/decrement of a channels prefetch_count. Arguments: callback (Callable): Function used to set new prefetch count, e.g. ``consumer.qos`` or ``channel.basic_qos``. Will be called with a single ``prefetch_count`` keyword argument. initial_value (int): Initial prefetch count value.. Example: >>> from kombu import Consumer, Connection >>> connection = Connection('amqp://') >>> consumer = Consumer(connection) >>> qos = QoS(consumer.qos, initial_prefetch_count=2) >>> qos.update() # set initial >>> qos.value 2 >>> def in_some_thread(): ... qos.increment_eventually() >>> def in_some_other_thread(): ... qos.decrement_eventually() >>> while 1: ... if qos.prev != qos.value: ... qos.update() # prefetch changed so update. It can be used with any function supporting a ``prefetch_count`` keyword argument:: >>> channel = connection.channel() >>> QoS(channel.basic_qos, 10) >>> def set_qos(prefetch_count): ... print('prefetch count now: %r' % (prefetch_count,)) >>> QoS(set_qos, 10) NcCs||_t|_|p d|_dSr/)callback threadingRLock_mutexvalue)r?r initial_valuer"r"r#r>s z QoS.__init__rcCsZ|j|jr|jt|d|_Wd|jSWd|jS1s%wY|jS)zIncrement the value, but do not update the channels QoS. Note: The MainThread will be responsible for calling :meth:`update` when necessary. rN)rrmaxr?nr"r"r#increment_eventuallys  zQoS.increment_eventuallycCsx|j.|jr|j|8_|jdkr(d|_Wd|jSWd|jSWd|jS1s4wY|jS)zDecrement the value, but do not update the channels QoS. Note: The MainThread will be responsible for calling :meth:`update` when necessary. rN)rrrr"r"r#decrement_eventuallys    zQoS.decrement_eventuallycCsH||jkr"|}|tkrtdtd}td||j|d||_|S)z#Set channel prefetch_count setting.z(QoS: Disabled: prefetch_count exceeds %rrzbasic.qos: prefetch_count->%s)prefetch_count)prevPREFETCH_COUNT_MAXrwarningdebugr)r?pcount new_valuer"r"r#sets   zQoS.setcCs6|j||jWdS1swYdS)z)Update prefetch count with current value.N)rrr)r?r"r"r#updates $z QoS.update)r) rDrErFrGrr>rrrrr"r"r"r#rds(   r)T)NF)rNN)NNF)NFNr)NN)BrG __future__rrr0rrr collectionsr contextlibr functoolsr itertoolsrrrr r r amqpr r rMrrfiverrlogr serializationrr} utils.uuid_threadr ImportErrorthread dummy_thread__all__rrDrr r$r.r3rrOrrVrQrPrrrrrrrrrrobjectrr"r"r"r#sl           ,   &    !