File: //usr/local/lib/python3.9/site-packages/kombu/__pycache__/messaging.cpython-39.pyc
a
X>h)b � @ s� d Z ddlmZ ddlmZ ddlmZ ddlmZ ddl m
Z
ddl m Z m
Z
mZ dd lmZmZmZ dd
lmZ dd lmZmZ dd lmZmZ er�dd
lm Z dZ G dd� d�Z G dd� d�Z dS )z Sending and receiving messages.� )� annotations)�count)�
TYPE_CHECKING� )�
maybe_declare)�compress)�PooledConnection�
is_connection�
maybe_channel)�Exchange�Queue�maybe_delivery_mode)�ContentDisallowed)�dumps�prepare_accept_content)�ChannelPromise�
maybe_list)�
TracebackType)r r �Producer�Consumerc @ s� e Zd ZdZdZdZdZdZdZdZ dZ
d.dd�Z dd� Z d d
� Z
d d � Zd
d� Zd/dd�Zdeefdd�Zd0dd�Zd1dd�Zdd� Zdd � Zeee�Zd d � Zd d � Zd!d"d#d$d%�d&d'�Z d(d)� Z e Z d2d*d+�Z ed,d-� �Z dS )3r a� Message Producer.
Arguments:
---------
channel (kombu.Connection, ChannelT): Connection or channel.
exchange (kombu.entity.Exchange, str): Optional default exchange.
routing_key (str): Optional default routing key.
serializer (str): Default serializer. Default is `"json"`.
compression (str): Default compression method.
Default is no compression.
auto_declare (bool): Automatically declare the default exchange
at instantiation. Default is :const:`True`.
on_return (Callable): Callback to call for undeliverable messages,
when the `mandatory` or `immediate` arguments to
:meth:`publish` is used. This callback needs the following
signature: `(exception, exchange, routing_key, message)`.
Note that the producer needs to drain events to use this feature.
N� Tc C sz || _ || _|p| j| _|p | j| _|p,| j| _|p8| j| _d | _| jd u rVtd�| _|d urd|| _| j rv| � | j � d S )Nr )
�_channel�exchange� routing_key�
serializer� compression� on_returnZ_channel_promiser � auto_declare�revive)�self�channelr r r r r r � r! �9/usr/local/lib/python3.9/site-packages/kombu/messaging.py�__init__A s
zProducer.__init__c C s d| j � d�S )Nz <Producer: �>�r �r r! r! r" �__repr__S s zProducer.__repr__c C s | j | �� fS �N)� __class__�__reduce_args__r& r! r! r" �
__reduce__V s zProducer.__reduce__c C s d | j | j| j| j| jfS r( )r r r r r r& r! r! r" r* Y s �zProducer.__reduce_args__c C s | j jr| j �� dS )z�Declare the exchange.
Note:
----
This happens automatically at instantiation when
the :attr:`auto_declare` flag is enabled.
N)r �name�declarer& r! r! r" r- ] s zProducer.declareFc K s |rt || j|fi |��S dS )z=Declare exchange if not already declared during this session.N)r r )r �entity�retry� retry_policyr! r! r" r h s zProducer.maybe_declarec C s2 t ||�r |j||p|j�fS |||p,| jj�fS r( )�
isinstancer, �
delivery_moder )r r r2 r
r r! r! r" �_delivery_detailsm s
�
�zProducer._delivery_detailsr c K s | j }|du rg n|}|
du r"i n|
}
|du r2i n|}|du rD| jn|}| du rV| jn| } | �| pf| j|�\}|d< |dur�tt|d ��|d< | �|| ||| |
�\}}}| jr�| jj r�| j|vr�|�
| j� |
r�| j j �
|� | j j| |fi |��}||||||
|||||||||
|�S )a� Publish message to the specified exchange.
Arguments:
---------
body (Any): Message body.
routing_key (str): Message routing key.
delivery_mode (enum): See :attr:`delivery_mode`.
mandatory (bool): Currently not supported.
immediate (bool): Currently not supported.
priority (int): Message priority. A number between 0 and 9.
content_type (str): Content type. Default is auto-detect.
content_encoding (str): Content encoding. Default is auto-detect.
serializer (str): Serializer to use. Default is auto-detect.
compression (str): Compression method to use. Default is none.
headers (Dict): Mapping of arbitrary headers to pass along
with the message body.
exchange (kombu.entity.Exchange, str): Override the exchange.
Note that this exchange must have been declared.
declare (Sequence[EntityT]): Optional list of required entities
that must have been declared before publishing the message.
The entities will be declared using
:func:`~kombu.common.maybe_declare`.
retry (bool): Retry publishing, or declaring entities if the
connection is lost.
retry_policy (Dict): Retry configuration, this is the keywords
supported by :meth:`~kombu.Connection.ensure`.
expiration (float): A TTL in seconds can be specified per message.
Default is no expiration.
timeout (float): Set timeout to wait maximum timeout second
for message to publish.
confirm_timeout (float): Set confirm timeout to wait maximum timeout second
for message to confirm publishing if the channel is set to confirm publish mode.
**properties (Any): Additional message properties, see AMQP spec.
Nr2 i� �
expiration)�_publishr r r3 r �str�int�_preparer r, �append�
connectionZtransport_options�updateZensure)r �bodyr r2 � mandatory� immediate�priority� content_type�content_encodingr �headersr r r/ r0 r- r4 �timeout�confirm_timeout�
propertiesr5 Z
exchange_namer! r! r" �publishz s4 )
� �
�zProducer.publishc C s� |d u r i n|}| j }|�||||||�}| rR| j}| D ]}||fd|i|�� q8|�d�}t|t�rp|j|d< |j||
||| | |
d�S )Nr/ �reply_to)r r r= r> rC rD )r Zprepare_messager �getr1 r r, Z
basic_publish)r r< r? r@ rA rB rE r r= r> r r- rC rD r/ r0 r �messager r. rG r! r! r" r5 � s& �
�zProducer._publishc C sD | j }t|t�r@|� }| _ | j�|� | jr@|jd �| j� |S )N� basic_return)r r1 r r r r �events�add�r r r! r! r" � _get_channel� s
zProducer._get_channelc C s
|| _ d S r( r% rM r! r! r" � _set_channel� s zProducer._set_channelc sp t |�r"|� � | _t� fdd��}t|t�r@|| _| �|�| _n,|| _| jr`| jjd �| j� | �|�| _dS )z*Revive the producer after connection loss.c s � j S r( )Zdefault_channelr! �r: r! r" �<lambda>� � z!Producer.revive.<locals>.<lambda>rJ N) r �__connection__r r1 r r r rK rL rM r! rP r" r � s
zProducer.revivec C s | S r( r! r& r! r! r" � __enter__� s zProducer.__enter__�type[BaseException] | None�BaseException | None�TracebackType | None�None��exc_type�exc_val�exc_tb�returnc C s: | j d ur.|d ur.t| j t�r.| j j�| j � | �� d S r( )rS r1 r �_pool�replace�release)r rZ r[ r\ r! r! r" �__exit__� s zProducer.__exit__c C s d S r( r! r& r! r! r" r` s zProducer.releasec C sh |s"|p | j }t||d�\}}}n&t|t�r@|s4d}|�|�}n|sHd}|r^t||�\}|d< |||fS )N)r zutf-8�binaryr )r r r1 r6 �encoder )r r< r r@ rA r rB r! r! r" r8 s
��
zProducer._preparec C s* z| j p| jjjW S ty$ Y n0 d S r( )rS r r: �client�AttributeErrorr& r! r! r" r: + s zProducer.connection)NNNNNN)F)NNFFr NNNNNNFNNNNN)NNFN)NNNNN)!�__name__�
__module__� __qualname__�__doc__r r r r r r rS r# r' r+ r* r- r r
r r3 rF r5 rN rO �propertyr r rT ra r` �closer8 r: r! r! r! r" r sV �
�
�
L �
�
r c @ s, e Zd ZdZeZdZdZdZdZdZ dZ
dZ dZ dZ
dZed�Zd9dd�Zedd� �Zejd d� �Zd
d � Zd d
� Zdd� Zdd� Zddddd�dd�Zdd� Zd:dd �Zd d � ZeZ d d � Z d!d"� Z d#d$� Z d%d&� Z d;d)d*�Z!d<d+d,�Z"d-d.� Z#dedfd/d0�Z$d=d1d2�Z%d3d4� Z&d5d6� Z'ed7d8� �Z(dS )>r a Message consumer.
Arguments:
---------
channel (kombu.Connection, ChannelT): see :attr:`channel`.
queues (Sequence[kombu.Queue]): see :attr:`queues`.
no_ack (bool): see :attr:`no_ack`.
auto_declare (bool): see :attr:`auto_declare`
callbacks (Sequence[Callable]): see :attr:`callbacks`.
on_message (Callable): See :attr:`on_message`
on_decode_error (Callable): see :attr:`on_decode_error`.
prefetch_count (int): see :attr:`prefetch_count`.
NTr c C s� || _ t|pg �| _|d u r"| jn|| _|d u r:| jp<g n|| _|| _|
| _i | _|d ur`|| _|d urn|| _ t
|�| _ | | _ | j r�| �
| j � d S r( )r r �queues�no_ack� callbacks�
on_message�
tag_prefix� _active_tagsr �on_decode_errorr �accept�prefetch_countr ) r r rl rm r rn rr ro rs rt rp r! r! r" r# � s" �
zConsumer.__init__c C s t | j�� �S r( )�list�_queues�valuesr& r! r! r" rl � s zConsumer.queuesc C s dd� |D �| _ d S )Nc S s i | ] }|j |�qS r! )r, )�.0�qr! r! r" �
<dictcomp>� rR z#Consumer.queues.<locals>.<dictcomp>)rv )r rl r! r! r" rl � s c C s� | j �� t|� }| _t| j�� �D ]6\}}| j�|d� || j� }| j|j< |� |� q&| j
rl| � � | j dur�| j
| j d� dS )z&Revive consumer after connection loss.N)rt )rq �clearr
r ru rv �items�popr, r r r- rt �qos)r r �qname�queuer! r! r" r � s
zConsumer.revivec C s | j �� D ] }|�� q
dS )z�Declare queues, exchanges and bindings.
Note:
----
This is done automatically at instantiation
when :attr:`auto_declare` is set.
N)rv rw r- �r r� r! r! r" r- � s zConsumer.declarec C s | j �|� dS )a% Register a new callback to be called when a message is received.
Note:
----
The signature of the callback needs to accept two arguments:
`(body, message)`, which is the decoded message body
and the :class:`~kombu.Message` instance.
N)rn r9 )r �callbackr! r! r" �register_callback� s zConsumer.register_callbackc C s | � � | S r( )�consumer& r! r! r" rT � s zConsumer.__enter__rU rV rW rX rY c C sH | j rD| j jrD| j jjj}t||�sDz | �� W n tyB Y n0 d S r( )r r: rd Zconnection_errorsr1 �cancel� Exception)r rZ r[ r\ Z conn_errorsr! r! r" ra � s
zConsumer.__exit__c C s( || j �}| jr|�� || j|j< |S )z�Add a queue to the list of queues to consume from.
Note:
----
This will not start consuming from the queue,
for that you will have to call :meth:`consume` after.
)r r r- rv r, r� r! r! r" � add_queue� s
zConsumer.add_queuec C sh t | j�� �}|rd|du r | jn|}|dd� |d }}|D ]}| j||dd� q>| j||dd� dS )aV Start consuming messages.
Can be called multiple times, but note that while it
will consume from new queues added since the last call,
it will not cancel consuming from removed queues (
use :meth:`cancel_by_queue`).
Arguments:
---------
no_ack (bool): See :attr:`no_ack`.
N���T�rm �nowaitF)ru rv rw rm �_basic_consume)r rm rl �H�Tr� r! r! r" r� � s zConsumer.consumec C s. | j j}| j�� D ] }||� q| j�� dS )z�End all active queue consumers.
Note:
----
This does not affect already delivered messages, but it does
mean the server will not send any more messages for this consumer.
N)r � basic_cancelrq rw r{ )r r� �tagr! r! r" r� s
zConsumer.cancelc
C sl t |t�r|jn|}zBz| j�|�}W n ty8 Y n0 | j�|� W | j�|d� n| j�|d� 0 dS )z Cancel consumer by queue name.N) r1 r r, rq r} �KeyErrorr r� rv )r r� r r� r! r! r" �cancel_by_queue s zConsumer.cancel_by_queuec C s |}t |t�r|j}|| jv S )z8Return :const:`True` if currently consuming from queue'.)r1 r r, rq )r r� r, r! r! r" �consuming_from s
zConsumer.consuming_fromc C s t dd� | j�� D ��S )z�Purge messages from all queues.
Warning:
-------
This will *delete all ready messages*, there is no undo operation.
c s s | ]}|� � V qd S r( )�purge)rx r� r! r! r" � <genexpr>- rR z!Consumer.purge.<locals>.<genexpr>)�sumrv rw r& r! r! r" r� & s zConsumer.purgec C s | j �|� dS )a� Enable/disable flow from peer.
This is a simple flow-control mechanism that a peer can use
to avoid overflowing its queues or otherwise finding itself
receiving more messages than it can process.
The peer that receives a request to stop sending content
will finish sending the current content (if any), and then wait
until flow is reactivated.
N)r �flow)r �activer! r! r" r� / s z
Consumer.flowr Fc C s | j �|||�S )a Specify quality of service.
The client can request that messages should be sent in
advance so that when the client finishes processing a message,
the following message is already held locally, rather than needing
to be sent down the channel. Prefetching gives a performance
improvement.
The prefetch window is Ignored if the :attr:`no_ack` option is set.
Arguments:
---------
prefetch_size (int): Specify the prefetch window in octets.
The server will send a message in advance if it is equal to
or smaller in size than the available prefetch size (and
also falls within other prefetch limits). May be set to zero,
meaning "no specific limit", although other prefetch limits
may still apply.
prefetch_count (int): Specify the prefetch window in terms of
whole messages.
apply_global (bool): Apply new settings globally on all channels.
)r Z basic_qos)r Z
prefetch_sizert Z apply_globalr! r! r" r~ < s �z Consumer.qosc C s | j j|d�S )a� Redeliver unacknowledged messages.
Asks the broker to redeliver all unacknowledged messages
on the specified channel.
Arguments:
---------
requeue (bool): By default the messages will be redelivered
to the original recipient. With `requeue` set to true, the
server will attempt to requeue the message, potentially then
delivering it to an alternative subscriber.
)�requeue)r Z
basic_recover)r r� r! r! r" �recoverY s
zConsumer.recoverc s* | j }|std��� �fdd�|D � dS )a� Method called when a message is received.
This dispatches to the registered :attr:`callbacks`.
Arguments:
---------
body (Any): The decoded message body.
message (~kombu.Message): The message instance.
Raises
------
NotImplementedError: If no consumer callbacks have been
registered.
z$Consumer does not have any callbacksc s g | ]}|� ���qS r! r! )rx r� �r< rI r! r" �
<listcomp>z rR z$Consumer.receive.<locals>.<listcomp>N)rn �NotImplementedError)r r< rI rn r! r� r" �receiveh s zConsumer.receivec C s: | j �|j�}|d u r6| �||�}|j|| j||d� |S )Nr� )rq rH r, �_add_tagr� �_receive_callback)r r� � consumer_tagrm r� r� r! r! r" r� | s
�zConsumer._basic_consumec C s( |pd� | jt| j��}|| j|j< |S )Nz{}{})�formatrp �next�_tagsrq r, )r r� r� r� r! r! r" r� � s
� zConsumer._add_tagc
C s� | j }| j| jd }}}zNt|dd �}|r4||�}|d urB||_ |jrV|�| j�W S |r^d n|�� }W n8 ty� } z | js�� | �||� W Y d }~n"d }~0 0 |r�||�S | � ||�S d S )NZmessage_to_python)
rs ro r �getattr�errorsZ_reraise_errorrr �decoder� r� )r rI rs Zon_mr �decodedZm2p�excr! r! r" r� � s "zConsumer._receive_callbackc C s dt | �j� d| j� d�S )N�<z: r$ )�typerf rl r& r! r! r" r' � s zConsumer.__repr__c C s$ z | j jjW S ty Y n0 d S r( )r r: rd re r& r! r! r" r: � s zConsumer.connection) NNNNNNNNN)N)r r F)F)N))rf rg rh ri r r rl rm r rn ro rr rs rt rv r r� r# rj �setterr r- r� rT ra r� r� r� rk r� r� r� r� r~ r� r� r� r� r� r' r: r! r! r! r" r 3 sZ �
�
r N) ri �
__future__r � itertoolsr �typingr �commonr r r r: r r r
r. r r r
�
exceptionsr Z
serializationr r Zutils.functionalr r �typesr �__all__r r r! r! r! r" �<module> s"