^f4dZddlmZddlZddlZddlZddlmZddlm 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-d Z0d!Z1d"Z2e d#Z3d.d$Z4d.d%Z5d/d&Z6Gd'd(Z7dS)0zCommon Utilities.) annotationsN)deque)contextmanager)partial)count) NAMESPACE_OIDuuid3uuid4uuid5) ChannelErrorRecoverableConnectionError)ExchangeQueue) get_logger)registry)uuid) Broadcast maybe_declarer itermessages send_replycollect_repliesinsureddrain_consumer eventloopicDttjatSN)_node_idr int>/var/www/html/env/lib/python3.11/site-packages/kombu/common.py get_node_idr#"s77; Or!cd|||t|} ttt|}n2#t $r%tt t|}YnwxYw|S)Nz{:x}-{:x}-{:x}-{:x})formatidstrr r ValueErrorr )node_id process_id thread_idinstanceentrets r" generate_oidr/)s  & &Y8  6 6C-% s++,, ---% s++,,- Js"A ,A98A9Tctttj|rt jnd|SNr)r/r#osgetpid threading get_ident)r,threadss r"oid_fromr73s=   !(/ a   r!cDeZdZdZejdzZ 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. ))queueNNFTc |r%d|pdt}n|pdt}tjd|p||||||nt |dd|dS)Nz{}.{}bcastzbcast.fanout)type)aliasr9name auto_deleteexchanger )r%rsuper__init__r) selfr?r9uniquer@rAr>kwargs __class__s r"rCzBroadcast.__init__Rs  /NN5#3GTVV<V  H  5  A()?@@@ NN7N### U K 4r!c t|||jjstd|jjjj|t fi|||S)Nr_)r]rSrOr rPensurerW)rRrSrYs r"rVrVswVW--- > $A()?@@@ 0 26> $ + 2 0 0". 0 006 A AAr!c#Ktfd}|g|pgz|_|5t|jjj||dD])} V#t$rY&wxYw ddddS#1swxYwYdS)z&Drain messages from consumer instance.c6||fdSr)append)bodymessageaccs r" on_messagez"drain_consumer..on_messages D'?#####r!T)limittimeoutignore_timeoutsN)r callbacksrrSrOrPpopleft IndexError)consumerrprqrsro_rns @r"rrs ''C$$$$$%b9H 8+6=!&OOO  A kkmm####      s5%B A#"B# A0-B/A00BBBc Ft|jd|g|d||||S)zIterator over messages.)queuesrS)rprqrsr )rConsumer)connrSr9rprqrsrFs r"rrs@  @eWg@@@@W    r!c#K|rt|p tD]5} ||V#tj$r|r|sY2wxYwdS)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|| .uww ##G#44 4 4 4 4~      s>AAc |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. )rArXrYreply_tocorrelation_id) routing_keyr serializercontent_encoding)publishdict propertiesget serializers type_to_name content_typer)rAreqmsgproducerrXrYpropss r"rrs 8   ,   s~j9"%."4"45E"F"F)6s7GH$'$8:: D D>C D D  r!c/&K|dd}d} t|||g|Ri|D]!\}}|s|d}|V" |r||jdSdS#|r||jwwxYw)z,Generator collecting replies from ``queue``.no_ackTFN) setdefaultrackafter_reply_message_receivedr?) r{rSr9argsrFrreceivedrlrms r"rrs   x . .FH =)$;+/;;;39;;  MD'  HJJJJ    =  0 0 < < < < < = =8 =  0 0 < < < < =s 4A11BcBtd||ddS)Nz#Connection error: %r. Retry in %ss T)exc_info)loggererror)excintervals r"_ensure_errbackrs1 LL.Xr!c#FK dVdS#|j|jz$rYdSwxYwr)connection_errorschannel_errors)r{s r"_ignore_errorsrsE    !D$7 7     s    c|r/t|5||i|cdddS#1swxYwYt|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)r{funrrFs r" ignore_errorsr%s8 ( D ! ! ( (3''' ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( $  s '++c$|r ||dSdSrr )rOrS on_revives r"revive_connectionrGs* 'r!c 8|pt}|d5}|||j}t t ||} |j||f|| d|} | |it||\} } | cdddS#1swxYwYdS)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"rrLs (G D ! ! T w///&*DIFFF $.g;w+1;;59;;GTCT&T%B%B%BCC                   sA"BBBc8eZdZdZdZdZd dZd dZdZdZ 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) NcV||_tj|_|pd|_dSr1)callbackr4RLock_mutexvalue)rDr initial_values r"rCz QoS.__init__s(  o'' "'a r!rc|j5|jr|jt|dz|_dddn #1swxYwY|jS)zIncrement the value, but do not update the channels QoS. Note: ---- The MainThread will be responsible for calling :meth:`update` when necessary. rN)rrmaxrDns r"increment_eventuallyzQoS.increment_eventuallys[ 4 4z 4!Z#a))3  4 4 4 4 4 4 4 4 4 4 4 4 4 4 4zs %9==c|j5|jr"|xj|zc_|jdkrd|_dddn #1swxYwY|jS)zDecrement the value, but do not update the channels QoS. Note: ---- The MainThread will be responsible for calling :meth:`update` when necessary. rN)rrrs r"decrement_eventuallyzQoS.decrement_eventuallys[ # #z # a :>>!"DJ  # # # # # # # # # # # # # # # zs*>AAc||jkrg|}|tkr"tdtd}td|||||_|S)z#Set channel prefetch_count setting.z(QoS: Disabled: prefetch_count exceeds %rrzbasic.qos: prefetch_count->%s)prefetch_count)prevPREFETCH_COUNT_MAXrwarningdebugr)rDpcount new_values r"setzQoS.setsv TY  I***I1333 LL8) D D D MMM 3 3 3DI r!cx|j5||jcdddS#1swxYwYdS)z)Update prefetch count with current value.N)rrr)rDs r"updatez QoS.updates [ ( (88DJ'' ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( (s /33)r) rHrIrJrKrrCrrrrr r!r"rr`s{((T 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"""""" %%%%%%33333333333399999999########222222   H   * * * * * * * * ZAAA++++    :AAA$9=$$$$P9=2 = = =        D (^(^(^(^(^(^(^(^(^(^(r!