o

'we�4�@
s�
dZd
dl
mZ
d
dlZd
dlZd
dlZd
dlmZ
d
dlm	Z	
d
dl
mZ
d
dlm
Z

d
dlmZmZmZmZ
d
d	lmZmZ
d
dlmZmZ
d
dlmZ
d
d
lmZ
d
dlmZ
dZdZ ee!�
Z"da#dd�Z$dd�Z%d;dd�
Z&Gdd�de�Z'dd�Z(d<dd�
Z)dd �Z*d!d"�Z+d#d$�Z,d=d%d&�
Z-		d=d'd(�
Z.d>d)d*�
Z/	d?d+d,�
Z0d-d.�Z1d/d0�Z2e	d1d2��
Z3d@d3d4�
Z4d@d5d6�
Z5dAd7d8�
Z6Gd9d:�d:�Z7dS)BzCommon Utilities.�)
�annotationsN)
�deque)
�contextmanager)
�partial)
�count)�
NAMESPACE_OID�uuid3�uuid4�uuid5)�ChannelError�RecoverableConnectionError�
)�Exchange�Queue)
�
get_logger)
�registry)
�uuid)	�	Broadcast�
maybe_declarer�itermessages�
send_reply�collect_replies�insured�drain_consumer�	eventloopi��cC
stdurt
�jatS�
N)�_node_idr	�int�rr�E/home/arjun/projects/env/lib/python3.10/site-packages/kombu/common.py�get_node_id"s

r cC
sLd
�||
|t
|�
�}z
ttt|��
}W|Sty%


ttt|��
}Y|Sw)Nz{:x}-{:x}-{:x}-{:x})�format�id�strrr�
ValueErrorr
)�node_id�
process_id�	thread_id�instance�ent�retrrr�generate_oid)s

�
�

�r+TcC
s$tt
�t��|
rt��|�Sd
|�S�Nr
)r+r �os�getpid�	threading�	get_ident)r(�threadsrrr�oid_from3s





�
�r2c
s8eZ
dZd
ZejdZ						d�f
dd�	Z�ZS)	ra�Broadcast 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
���}t�jd|p|
||||du
r$|nt|
dd�d�|�
�

dS)Nz{}.{}�bcastzbcast.�fanout)
�type)�aliasr3�name�auto_delete�exchanger)r!r�super�__init__r)�selfr8r3�uniquer9r:r7�kwargs�
�	__class__rrr<Rs








�
�zBroadcast.__init__)NNFTNN)�__name__�
__module__�__qualname__�__doc__r�attrsr<�
__classcell__rrr@rr<s






�rcC
s||
jj
jvSr)�
connection�client�declared_entities)�entity�channelrrr�declaration_cachedis
rMFcK
s |rt||
fi|�
�
St
||
�S)
zDeclare entity (cached).)�_imaybe_declare�_maybe_declare)rKrL�retry�retry_policyrrrrms


rcC
s4|j}|s|
st
d
|
�d|���
�
|�|
�
}|SdS)z�Make 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



�

�rTcC
s�|}t||
�
|
dur|j
std
|�d��
�
|j}
d}}|
jr2|jr2|
jjj}t|�
}||vr2dS|
js9t	d�
�
|j
|
d�

|du
rJ|rJ|�|�

|du
rR|j|_dS)Nzchannel is None and entity z not bound.F�channel disconnected)
rLT)
rTrRrrLrH�can_cache_declarationrIrJ�hashr�declare�addr8)rKrL�orig�declared�identrrrrO�s,




�













rOcK
s:t||
�
|j
js
td
�
�
|j
jjj|tfi|�
�
||
�S)NrU)rTrLrHrrI�ensurerO)rKrLrQrrrrN�s




�
�
�rNc
#
s��t���f
d
d�}|g
|pg|_
|�'
t|jjj|
|dd�D]}z���V
Wq ty2


Yq wWd�
dS1s>w



Y
dS)z&Drain messages from consumer instance.c
s��||
f�

dSr)
�append)�body�message�
�accrr�
on_message�s
z"drain_consumer.<locals>.on_messageT)�limit�timeout�ignore_timeoutsN)r�	callbacksrrLrHrI�popleft�
IndexError)�consumerrdrergrc�
_rrarr�s�



�


��"�rcK
s$t|j
d|g
|
d
�|�
�
|||d�S)zIterator over messages.)�queuesrL)rdrergNr)r�Consumer)�connrLr3rdrergr?rrrr�s

�rc	c
sN�|
rt|
�
p	t
�D]}z	|j|d
�
V
Wq
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_events�socketre)rnrdrerf�
irrrr�s�




���rc	K
sH|j|f
|||d
�t
|
jd|
j�d�
tj|
j|
jd�f
i|�
�
�
�
S)a�
Send 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:rPrQ�reply_to�correlation_id)�routing_keyrt�
serializer�content_encoding)�publish�dict�
properties�get�serializers�type_to_name�content_typerw)r:�req�msg�producerrPrQ�propsrrrr�s
�

�





���rc		o
s|�|�d
d�}d}z*t
||
|g|�
Ri|�
�
D]\}}|s!|��
d}|V
qW|r2|
�|j�

dSdS|r=|
�|j�

ww)z,Generator collecting replies from ``queue``.�no_ackTFN)�
setdefaultr�ack�after_reply_message_receivedr8)	rnrLr3�argsr?r��receivedr_r`rrrr
s&�




�
�


�
�
�rcC
stj
d
||
dd�
dS)Nz#Connection error: %r. Retry in %ss
T)
�exc_info)�logger�error)�exc�intervalrrr�_ensure_errback
s



�r�c

	c
s,�zdV
WdS|j|j
y


YdSwr)�connection_errors�channel_errors)
rnrrr�_ignore_errors
s�


�r�cO
sB|
rt|�
�
|
|i|�
�
Wd
�
S1sw



Y
t|�
S)a�Ignore 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�)rn�funr�r?rrr�
ignore_errors%
s



 �r�cC
s|r||
�

dSdSrr)rHrL�	on_reviverrr�revive_connectionG
s

�r�c
K
s�|pt}|j
d
d�
�4}|j|d�

|j}tt||d�}	|j|
|f||	d�|�
�
}
|
|it||d��
�
\}}|Wd�
S1sAw



Y
dS)z�Function wrapper to handle connection errors.

    Ensures function performing broker commands completes
    despite intermittent connection failures.
    T)
�block)
�errback)
r�)r�r�)
rHN)r��acquire�ensure_connection�default_channelrr��	autoretryry)
�poolr�r�r?r�r��optsrnrL�reviver�retvalrkrrrrL
s



�
�
$�rc@
s@eZ
dZd
ZdZdd�Zddd�
Zddd	�
Zd
d�Zdd
�Z	dS)�QoSa�Thread 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)
    NcC
s|
|_t
��|_|pd
|_dSr,)�callbackr/�RLock�_mutex�value)r=r��
initial_valuerrrr<�
s



zQoS.__init__r
cC
sZ|j�
|j
r|j
t|
d
�|_
Wd�
|j
SWd�
|j
S1s%w



Y
|j
S)z�Increment the value, but do not update the channels QoS.

        Note:
        ----
            The MainThread will be responsible for calling :meth:`update`
            when necessary.
        r
N)r�r��max�r=�
nrrr�increment_eventually�
s


��
��zQoS.increment_eventuallycC
sx|j�.
|j
r|j
|
8_
|j
d
kr(d
|_
Wd�
|j
SWd�
|j
SWd�
|j
S1s4w



Y
|j
S)z�Decrement the value, but do not update the channels QoS.

        Note:
        ----
            The MainThread will be responsible for calling :meth:`update`
            when necessary.
        r
N)r�r�r�rrr�decrement_eventually�
s





��
��
��zQoS.decrement_eventuallycC
sH|
|jkr"|
}|
t
krt�d
t
�
d}t�d|�
|j|d�

|
|_|
S)z#Set channel prefetch_count setting.z(QoS: Disabled: prefetch_count exceeds %rr
zbasic.qos: prefetch_count->%s)
�prefetch_count)�prev�PREFETCH_COUNT_MAXr��warning�debugr�)r=�pcount�	new_valuerrr�set�
s




�



zQoS.setc

C
s6|j�
|�
|j�
Wd
�
S1sw



Y
d
S)z)Update prefetch count with current value.N)r�r�r�)
r=rrr�update�
s

$�z
QoS.update)
r
)
rBrCrDrEr�r<r�r�r�r�rrrrr�`
s
*



r�)
T)NF)r
NN)NNF)NFNr)NN)8rE�
__future__rr-rqr/�collectionsr�
contextlibr�	functoolsr�	itertoolsrrrrr	r
�amqprrrKrr�logr�
serializationrr|�
utils.uuid�__all__r�rBr�rr r+r2rrMrrTrOrNrrrrrr�r�r�r�rr�rrrr�<module>
sV











	-




�
	(
�



"