Qj7dZddlmZddlZddlZddlZddlmZddlm Z ddl m Z ddl m Z ddlmZmZmZmZdd lmZmZd d lmZmZd d lmZd d lmZd dlmZdZdZ ee!Z"da#dZ$dZ%d'dZ&GddeZ'dZ(d(dZ)dZ*dZ+dZ,d)dZ- d)dZ.d*dZ/ d+dZ0dZ1d Z2e d!Z3d,d"Z4d,d#Z5d-d$Z6Gd%d&Z7y).zCommon Utilities.) annotationsN)deque)contextmanager)partial)count) NAMESPACE_OIDuuid3uuid4uuid5) ChannelErrorRecoverableConnectionError)ExchangeQueue) get_logger)registry)uuid) Broadcast maybe_declarer itermessages send_replycollect_repliesinsureddrain_consumer eventloopicBttjatSN)_node_idr intH/var/www/html/truck-me/venv/lib/python3.12/site-packages/kombu/common.py get_node_idr#"s7;; Or!cdj|||t|} ttt|}|S#t $rtt t|}Y|SwxYw)Nz{:x}-{:x}-{:x}-{:x})formatidstrr r ValueErrorr )node_id process_id thread_idinstanceentrets r" generate_oidr/)sc  & &Y8  6C-% s+, J -% s+, J-s:"A A ctttj|rt j |Sd|SNr)r/r#osgetpid threading get_ident)r,threadss r"oid_fromr73s@   !(   /0  r!cNeZdZdZej dzZ dfd ZxZS)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. ))queueNc |rdj|xsdt}n|xs dt}t| d|xs||||||n t |dd|y)Nz{}.{}bcastzbcast.fanout)type)aliasr9name auto_deleteexchanger )r%rsuper__init__r) selfr?r9uniquer@rAr>kwargs __class__s r"rCzBroadcast.__init__Rsp NN5#3GTV>V  H    ()?@@ NN7N# U KK r!c t||}|jjs td|jjjj |t fi|||S)Nr_)r]rSrOr rPensurerW)rRrSrYs r"rVrVsj %fg 6F >> $ $()?@@ 0 26>> $ $ + + 2 2 0". 006 AAr!c#"Ktfd}|g|xsgz|_|5t|jjj ||dD]} j  dddy#t$rY-wxYw#1swYyxYww)z&Drain messages from consumer instance.c,j||fyr)append)bodymessageaccs r" on_messagez"drain_consumer..on_messages D'?#r!T)limittimeoutignore_timeoutsN)r callbacksrrSrOrPpopleft IndexError)consumerrprqrsro_rns @r"rrs 'C$%b9H 8++66==!&O A kkm#     s@!B1BA4(B+ B4 B=B?BBB Bc Ht|jd|g|d||||S)zIterator over messages.)queuesrS)rprqrsr )rConsumer)connrSr9rprqrsrFs r"rrs2  @eWg@@W  r!c#K|xr t|xs tD]} |j|y#tj$r|r|sY5wxYww)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. )rqN)ranger drain_eventssocketrq)r{rprqrris r"rrs]> #uU| .uw ##G#4 4~~  s%A9AAAAAc |j|f|||dt|jd|jjdtj |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. )rArXrYreply_tocorrelation_id) routing_keyr serializercontent_encoding)publishdict propertiesget serializers type_to_name content_typer)rAreqmsgproducerrXrYpropss r"rrs 8    ,  s~~j9"%.."4"45E"F)66s7G7GH$'$8$8: D>C D r!c/K|jdd}d} t|||g|i|D]\}}|s|jd}| |r|j|jyy#|r|j|jwwxYww)z,Generator collecting replies from ``queue``.no_ackTFN) setdefaultrackafter_reply_message_receivedr?) r{rSr9argsrFrreceivedrlrms r"rrs   x .FH =)$;+/;39; MD' HJ     0 0 < 8  0 0 < sB 1A) B ) B  B c6tjd||dy)Nz#Connection error: %r. Retry in %ss T)exc_info)loggererror)excintervals r"_ensure_errbackrs LL.Xr!c#ZK dy#|j|jz$rYywxYwwr)connection_errorschannel_errors)r{s r"_ignore_errorsrs0   ! !D$7$7 7   s+ +(+(+c|rt|5||i|cdddSt|S#1swYt|SxYw)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)r{funrrFs r" ignore_errorsr'sG8 D ! ('' ( ( $  ( $ s+>c|r ||yyrr )rOrS on_revives r"revive_connectionrIs'r!c $|xst}|jd5}|j||j}t t ||} |j ||f|| d|} | |it||\} } | cdddS#1swYyxYw)zFunction wrapper to handle connection errors. Ensures function performing broker commands completes despite intermittent connection failures. T)block)errback)r)rr)rON)racquireensure_connectiondefault_channelrr autoretryr) poolrrrFrroptsr{rSreviverretvalrws r"rrNs (G D ! T w/&&*DIF $..g;w+1;59;TCT&T%BC    s ABBc8eZdZdZdZddZd dZd dZdZdZ y) 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.. max_prefetch (int or None): Maximum allowed prefetch count. If specified as an integer, increment_eventually will not allow the value to exceed this limit. If None (the default), there is no upper limit on the prefetch count. 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) Nch||_tj|_|xsd|_||_yr1)callbackr4RLock_mutexvalue max_prefetch)rDr initial_valuers r"rCz QoS.__init__s+  oo' "'a (r!c|j5|jrG|jt|dz}|j||jkDr |j}||_ddd|jS#1swY|jSxYw)a Increment the value, but do not update the channels QoS. Note: ---- The MainThread will be responsible for calling :meth:`update` when necessary. If max_prefetch is set, the value will not exceed this limit. rN)rrmaxr)rDn new_values r"increment_eventuallyzQoS.increment_eventuallysy[[ 'zz JJQ2 $$0YARAR5R $ 1 1I&  ' zz  ' zzs AA55B c|j5|jr+|xj|zc_|jdkrd|_ddd|jS#1swY|jSxYw)zDecrement the value, but do not update the channels QoS. Note: ---- The MainThread will be responsible for calling :meth:`update` when necessary. rN)rr)rDrs r"decrement_eventuallyzQoS.decrement_eventuallysY[[ #zz a ::>!"DJ  # zz  # zzs 8AA-c||jk7rV|}|tkDrtjdtd}tj d||j |||_|S)z#Set channel prefetch_count setting.z(QoS: Disabled: prefetch_count exceeds %rrzbasic.qos: prefetch_count->%s)prefetch_count)prevPREFETCH_COUNT_MAXrwarningdebugr)rDpcountrs r"setzQoS.sets\ TYY I**I13 LL8) D MMM 3DI r!c||j5|j|jcdddS#1swYyxYw)z)Update prefetch count with current value.N)rrr)rDs r"updatez QoS.updates. [[ (88DJJ' ( ( (s2;r)r) rHrIrJrKrrCrrrrr r!r"rrbs(+Z D) "  (r!r)T)NF)rNN)NNF)NFNr)NN)8rK __future__rr2rr4 collectionsr contextlibr functoolsr itertoolsrrrr r r amqpr r rRrrlogr serializationrr utils.uuid__all__rrHrrr#r/r7rrTrr]rWrVrrrrrrrrrrrr r!r"rs" %339#2   H  * * ZA+  >A$9=$P9=2 =    D (f(f(r!