o ?ZcD4@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-Z/d.d/Z0ed0d1Z1d?d2d3Z2d?d4d5Z3d@d6d7Z4Gd8d9d9Z5dS)AzCommon Utilities.N)deque)contextmanager)partial)count) NAMESPACE_OIDuuid3uuid4uuid5) ChannelErrorRecoverableConnectionError)ExchangeQueue) get_logger)registry)uuid) Broadcast maybe_declarer itermessages send_replycollect_repliesinsureddrain_consumer eventloopicCstdurtjatSN)_node_idrintrrB/var/www/chikooza/env/lib/python3.10/site-packages/kombu/common.py get_node_id srcCsLd|||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*rosgetpid threading get_ident)r'threadsrrroid_from1s r1cs8eZdZdZejdZ      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)aliasr2name auto_deleteexchanger)r rsuper__init__r )selfr7r2uniquer8r9r6kwargs __class__rrr;Os   zBroadcast.__init__)NNFTNN)__name__ __module__ __qualname____doc__rattrsr; __classcell__rrr?rr:s rcCs||jjjvSr) connectionclientdeclared_entities)entitychannelrrrdeclaration_cachedfsrLFcKs |r t||fi|St||S)zDeclare entity (cached).)_imaybe_declare_maybe_declare)rJrKretry retry_policyrrrrjs 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)rJrKrQrrr_ensure_channel_is_boundqs rScCs|}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)rKT) rSrQr rKrGcan_cache_declarationrHrIhashr declareaddr7)rJrKorigdeclaredidentrrrrNs,       rNcKs:t|||jjs td|jjjj|tfi|||S)NrT)rSrKrGr rHensurerN)rJrKrPrrrrMs  rMc #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 callbacksrrKrGrHpopleft IndexError)consumerrcrdrfrb_rr`rrs    "rcKs$t|jd|g|d||||dS)zIterator over messages.)queuesrK)rcrdrfNr)rConsumer)connrKr2rcrdrfr>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. )rdN)ranger drain_eventssocketrd)rmrcrdreirrrrsrc 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. )r9rOrPreply_tocorrelation_id) routing_keyrs serializercontent_encoding)publishdict propertiesget serializers type_to_name content_typerv)r9reqmsgproducerrOrPpropsrrrrs   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_receivedr7) rmrKr2argsr>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)rmrrr_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)rmfunrr>rrr ignore_errorss    rcCs|r||dSdSrr)rGrK on_reviverrrrevive_connection@s 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)rGN)racquireensure_connectiondefault_channelrr autoretryrx) poolrrr>rroptsrmrKreviverretvalrjrrrrEs  $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 ) rArBrCrDrr;rrrrrrrrrYs(   r)T)NF)r NN)NNF)NFNr)NN)6rDr,rpr. collectionsr contextlibr functoolsr itertoolsrrrrrr amqpr r rJr rlogr serializationrr{ utils.uuid__all__rrArrrr*r1rrLrrSrNrMrrrrrrrrrrrrrrrsT         ,   &    !