U dD4@sdZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z mZmZmZddlmZmZd d lmZmZd d lmZd d lmZd d lm Z dZdZeeZ da!ddZ"ddZ#d:ddZ$GdddeZ%ddZ&d;ddZ'ddZ(d d!Z)d"d#Z*dd(d)Z-d?d*d+Z.d,d-Z/d.d/Z0ed0d1Z1d@d2d3Z2dAd4d5Z3dBd6d7Z4Gd8d9d9Z5dS)CzCommon Utilities.N)deque)contextmanager)partial)count) NAMESPACE_OIDuuid3uuid4uuid5) ChannelErrorRecoverableConnectionError)ExchangeQueue) get_logger)registry)uuid) Broadcast maybe_declarer itermessages send_replycollect_repliesinsureddrain_consumer eventloopicCstdkrtjatSN)_node_idrintrr0/tmp/pip-unpacked-wheel-hqfrjlvz/kombu/common.py get_node_id srcCsNd|||t|}zttt|}Wn"tk rHttt|}YnX|S)Nz{:x}-{:x}-{:x}-{:x})formatidstrrr ValueErrorr )Znode_idZ process_idZ thread_idinstanceentretrrr generate_oid'sr'TcCs ttt|rtnd|SNr)r'rosgetpid threading get_ident)r$threadsrrroid_from1s r.cs,eZdZdZejdZdfdd ZZS) raBroadcast queue. Convenience class used to define broadcast queues. Every queue instance will have a unique name, and both the queue and exchange is configured with auto deletion. Arguments: name (str): This is used as the name of the exchange. queue (str): By default a unique id is used for the queue name for every consumer. You can specify a custom queue name here. unique (bool): Always create a unique queue even if a queue name is supplied. **kwargs (Any): See :class:`~kombu.Queue` for a list of additional keyword arguments supported. ))queueNNFTc sb|rd|pdt}n|p&dt}tjf|p6|||||dk rH|n t|ddd|dS)Nz{}.{}Zbcastzbcast.Zfanout)type)aliasr/name auto_deleteexchange)r rsuper__init__r )selfr2r/uniquer3r4r1kwargs __class__rrr6Os  zBroadcast.__init__)NNFTNN)__name__ __module__ __qualname____doc__rattrsr6 __classcell__rrr:rr:s rcCs||jjjkSr) connectionclientdeclared_entities)entitychannelrrrdeclaration_cachedfsrGFcKs|rt||f|St||S)zDeclare entity (cached).)_imaybe_declare_maybe_declare)rErFretry retry_policyrrrrjsrcCs4|j}|s0|s"td|d|||}|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 zCannot bind channel z to entity N)is_boundr bind)rErFrLrrr_ensure_channel_is_boundqs rNcCs|}t|||dkr2|js,td|d|j}d}}|jrd|jrd|jjj}t|}||krddS|jsrt d|j |d|dk r|r| ||dk r|j |_ dS)Nzchannel is None and entity z not bound.Fchannel disconnected)rFT) rNrLr rFrBZcan_cache_declarationrCrDhashr Zdeclareaddr2)rErForigZdeclaredidentrrrrIs,       rIcKs6t|||jjstd|jjjj|tf|||S)NrO)rNrFrBr rCZensurerI)rErFrKrrrrHs  rHc #svtfdd}|g|pg|_|Ft|jjj||ddD](}zVWq>tk rdYq>Xq>W5QRXdS)z&Drain messages from consumer instance.cs||fdSr)append)bodymessageaccrr on_messagesz"drain_consumer..on_messageT)limittimeoutignore_timeoutsN)r callbacksrrFrBrCpopleft IndexError)ZconsumerrZr[r]rY_rrWrrs   rcKs$t|jf|g|d||||dS)zIterator over messages.)ZqueuesrF)rZr[r])rZConsumer)connrFr/rZr[r]r9rrrrs rc csP|r t|ptD]8}z|j|dVWqtjk rH|rD|sDYqXqdS)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. )r[N)rangerZ drain_eventssocketr[)rarZr[r\irrrrs rc KsD|j|f|||dt|jd|jdtj|j|jdf|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. )r4rJrKZreply_tocorrelation_id)Z routing_keyre serializercontent_encoding)publishdictZ propertiesget serializersZ type_to_name content_typerg)r4reqmsgZproducerrJrKpropsrrrrs   rc os`|dd}d}z8t|||f||D]\}}|s:|d}|Vq&W5|rZ||jXdS)z,Generator collecting replies from ``queue``.no_ackTFN) setdefaultZafter_reply_message_receivedr2rZack) rarFr/argsr9rpZreceivedrUrVrrrrs    rcCstjd||dddS)Nz#Connection error: %r. Retry in %ss T)exc_info)loggererror)excintervalrrr_ensure_errbacks rxc cs,z dVWn|j|jk r&YnXdSr)Zconnection_errorsZchannel_errors)rarrr_ignore_errorss ryc Os2|r*t||||W5QRSQRXt|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. )ry)rafunrrr9rrr ignore_errorss r{cCs|r ||dSrr)rBrF on_reviverrrrevive_connection@sr}c Ks|pt}|jddb}|j|d|j}tt||d} |j||f|| d|} | |t||d\} } | W5QRSQRXdS)zFunction wrapper to handle connection errors. Ensures function performing broker commands completes despite intermittent connection failures. T)block)errback)r|)rr|)rBN)rxacquireZensure_connectionZdefault_channelrr}Z autoretryri) poolrzrrr9rr|optsrarFZreviverretvalr`rrrrEs  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|_|pd|_dSr()callbackr+RLock_mutexvalue)r7r initial_valuerrrr6s z QoS.__init__r c Cs0|j|jr |jt|d|_W5QRX|jS)zIncrement the value, but do not update the channels QoS. Note: The MainThread will be responsible for calling :meth:`update` when necessary. r)rrmaxr7nrrrincrement_eventuallyszQoS.increment_eventuallyc Cs<|j*|jr,|j|8_|jdkr,d|_W5QRX|jS)zDecrement the value, but do not update the channels QoS. Note: The MainThread will be responsible for calling :meth:`update` when necessary. r )rrrrrrdecrement_eventuallys  zQoS.decrement_eventuallycCsH||jkrD|}|tkr&tdtd}td||j|d||_|S)z#Set channel prefetch_count setting.z(QoS: Disabled: prefetch_count exceeds %rrzbasic.qos: prefetch_count->%s)Zprefetch_count)prevPREFETCH_COUNT_MAXrtwarningdebugr)r7Zpcount new_valuerrrsets   zQoS.setc Cs*|j||jW5QRSQRXdS)z)Update prefetch count with current value.N)rrr)r7rrrupdatesz QoS.update)r )r ) r<r=r>r?rr6rrrrrrrrrYs(  r)T)NF)r NN)r NN)NNF)NFN)N)N)NN)6r?r)rcr+ collectionsr contextlibr functoolsr itertoolsrrrrrr Zamqpr r rEr rlogrZ serializationrrkZ utils.uuid__all__rr<rtrrr'r.rrGrrNrIrHrrrrrrxryr{r}rrrrrrsV        ,   &   !