hD4dZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z mZmZmZddlmZmZd d lmZmZd d lmZd d lmZd d lm Z dZdZeeZ da!dZ"dZ#d(dZ$GddeZ%dZ&d)dZ'dZ(dZ)dZ*d*dZ+ d*dZ,d+dZ- d,dZ.d Z/d!Z0ed"Z1d-d#Z2d-d$Z3d.d%Z4Gd&d'Z5dS)/zCommon Utilities.N)deque)contextmanager)partial)count) NAMESPACE_OIDuuid3uuid4uuid5) ChannelErrorRecoverableConnectionError)ExchangeQueue) get_logger)registry)uuid) Broadcast maybe_declarer itermessages send_replycollect_repliesinsureddrain_consumer eventloopicDttjatSN)_node_idr intH/var/www/html/Sam_Eipo/venv/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})formatidstrrr 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_fromr61s=   !(/ 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)aliasr8name auto_deleteexchanger)r$rsuper__init__r) selfr>r8uniquer?r@r=kwargs __class__s r!rBzBroadcast.__init__Os  /NN5#3GTVV<)rQrRorigdeclaredidents r!rVrVs DVW--- CAfAAACC C.Huf:%,>V  H  5  A()?@@@ NN7N### U K 4r c t|||jjstd|jjjj|t fi|||S)Nr^)r\rRrNr rOensurerV)rQrRrXs r!rUrUswVW--- > $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 callbacksrrRrNrOpopleft IndexError)consumerrorprrrn_rms @r!rrs ''C$$$$$%b9H 8+6=!&OOO  A kkmm####      s5%B A#"B# A0-B/A00BBBc Ft|jd|g|d||||S)zIterator over messages.)queuesrR)rorprrr)rConsumer)connrRr8rorprrrEs 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. )rpN)ranger drain_eventssocketrp)rzrorprqis 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. )r@rWrXreply_tocorrelation_id) routing_keyr serializercontent_encoding)publishdict propertiesget serializers type_to_name content_typer)r@reqmsgproducerrWrXpropss 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>) rzrRr8argsrErreceivedrkrls 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)rzs 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)rzfunrrEs r! ignore_errorsrs6 ( D ! ! ( (3''' ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( $  s '++c$|r ||dSdSrr)rNrR on_revives r!revive_connectionr@s* '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)rNN)racquireensure_connectiondefault_channelrr autoretryr) poolrrrErroptsrzrRreviverretvalrvs r!rrEs (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|_dSr0)callbackr3RLock_mutexvalue)rCr initial_values r!rBz QoS.__init__s(  o'' "'a r r c|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)rrmaxrCns 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. r N)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)rCpcount 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)rCs r!updatez QoS.updates [ ( (88DJ'' ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( ( (s /33)r ) rGrHrIrJrrBrrrrrr r!rrYs{&&P D(((           (((((r r)T)NF)r NN)NNF)NFNr)NN)6rJr1r~r3 collectionsr contextlibr functoolsr itertoolsrrrrr r amqpr r rQrrlogr serializationrr utils.uuid__all__rrGrrr"r.r6rrSrr\rVrUrrrrrrrrrrrrr r!rs %%%%%%33333333333399999999########222222   H   ) ) ) ) ) ) ) ) XAAA++++    :AAA$9=""""L9=0 = = =        B (Z(Z(Z(Z(Z(Z(Z(Z(Z(Z(r