o Df4@sdZddlmZddlZddlZddlZddlmZddlm Z ddl m Z ddl m Z ddlmZmZmZmZdd lmZmZd d lmZmZd d lmZd d lmZd dlmZdZdZ ee!Z"da#ddZ$ddZ%d;ddZ&GdddeZ'ddZ(dd)d*Z/ d?d+d,Z0d-d.Z1d/d0Z2e d1d2Z3d@d3d4Z4d@d5d6Z5dAd7d8Z6Gd9d:d:Z7dS)BzCommon Utilities.) annotationsN)deque)contextmanager)partial)count) NAMESPACE_OIDuuid3uuid4uuid5) ChannelErrorRecoverableConnectionError)ExchangeQueue) get_logger)registry)uuid) Broadcast maybe_declarer itermessages send_replycollect_repliesinsureddrain_consumer eventloopicCstdurtjatSN)_node_idr intrrE/home/ubuntu/webapp/venv/lib/python3.10/site-packages/kombu/common.py get_node_id"sr cCsLd|||t|}z ttt|}W|Sty%ttt|}Y|Sw)Nz{:x}-{:x}-{:x}-{:x})formatidstrrr ValueErrorr )node_id process_id thread_idinstanceentretrrr generate_oid)s  r+TcCs$ttt|rt|Sd|SNr)r+r osgetpid threading get_ident)r(threadsrrroid_from3s r2cs8eZdZdZejdZ      dfdd 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|r d|pdt}n|pdt}tjd|p|||||dur$|nt|ddd|dS)Nz{}.{}bcastzbcast.fanout)type)aliasr3name auto_deleteexchanger)r!rsuper__init__r)selfr8r3uniquer9r:r7kwargs __class__rrr<Rs   zBroadcast.__init__)NNFTNN)__name__ __module__ __qualname____doc__rattrsr< __classcell__rrr@rr<s rcCs||jjjvSr) connectionclientdeclared_entities)entitychannelrrrdeclaration_cachedisrMFcKs |r t||fi|St||S)zDeclare entity (cached).)_imaybe_declare_maybe_declare)rKrLretry retry_policyrrrrms rcCs4|j}|s|std|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)rKrLrRrrr_ensure_channel_is_boundts rTcCs|}t|||dur|jstd|d|j}d}}|jr2|jr2|jjj}t|}||vr2dS|js9t d|j |d|durJ|rJ| ||durR|j |_ dS)Nzchannel is None and entity z not bound.Fchannel disconnected)rLT) rTrRr rLrHcan_cache_declarationrIrJhashr declareaddr8)rKrLorigdeclaredidentrrrrOs,       rOcKs:t|||jjs td|jjjj|tfi|||S)NrU)rTrLrHr rIensurerO)rKrLrQrrrrNs  rNc #stfdd}|g|pg|_|'t|jjj||ddD]}zVWq ty2Yq wWddS1s>wYdS)z&Drain messages from consumer instance.cs||fdSr)append)bodymessageaccrr on_messagesz"drain_consumer..on_messageT)limittimeoutignore_timeoutsN)r callbacksrrLrHrIpopleft IndexError)consumerrdrergrc_rrarrs    "rcKs$t|jd|g|d||||dS)zIterator over messages.)queuesrL)rdrergNr)rConsumer)connrLr3rdrergr?rrrrsrc 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. )reN)ranger drain_eventssocketre)rnrdrerfirrrrsrc 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:rPrQreply_tocorrelation_id) routing_keyrt serializercontent_encoding)publishdict propertiesget serializers type_to_name content_typerw)r:reqmsgproducerrPrQpropsrrrrs   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_receivedr8) rnrLr3argsr?rreceivedr_r`rrrrs&  rcCstjd||dddS)Nz#Connection error: %r. Retry in %ss T)exc_info)loggererror)excintervalrrr_ensure_errbacks rc cs,zdVWdS|j|jyYdSwr)connection_errorschannel_errors)rnrrr_ignore_errorss  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)rnfunrr?rrr ignore_errors%s    rcCs|r||dSdSrr)rHrL on_reviverrrrevive_connectionGs 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)rHN)racquireensure_connectiondefault_channelrr autoretryry) poolrrr?rroptsrnrLreviverretvalrkrrrrLs  $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,)callbackr/RLock_mutexvalue)r=r initial_valuerrrr<s z QoS.__init__r cCsZ|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=nrrrincrement_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. r N)rrrrrrdecrement_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_valuerrrsets   zQoS.setcCs6|j||jWdS1swYdS)z)Update prefetch count with current value.N)rrr)r=rrrupdates $z QoS.update)r ) rBrCrDrErr<rrrrrrrrr`s*   r)T)NF)r NN)NNF)NFNr)NN)8rE __future__rr-rqr/ collectionsr contextlibr functoolsr itertoolsrrrrr r amqpr r rKrrlogr serializationrr| utils.uuid__all__rrBrrr r+r2rrMrrTrOrNrrrrrrrrrrrrrrrsV          -   (    "