File: //proc/self/root/usr/local/lib/python3.9/site-packages/kombu/__pycache__/compat.cpython-39.pyc
a
X>h` � @ s� d Z ddlmZ ddlmZ ddlmZ ddlmZ ddl m
Z
m Z erTddl m
Z
d Ze jZdd d
�ZG dd� dej�ZG dd� dej�ZG dd� dej�Zd S )zYCarrot compatibility interface.
See https://pypi.org/project/carrot/ for documentation.
� )� annotations)�count)�
TYPE_CHECKING� )� messaging)�Exchange�Queue)�
TracebackType)� Publisher�ConsumerFNc c s8 |j |d� td�D ] }|r(||kr( q4| �� V qd S )N)�no_ackr )Zconsumer Z drain_events)�
connection�consumerr �limit� iteration� r �6/usr/local/lib/python3.9/site-packages/kombu/compat.py� _iterconsume s
r c sv e Zd ZdZdZdZdZdZdZdZ d� fdd� Z
d d
� Z � fd d �Z d
d� Z
ddddd�dd�Zedd� �Z� ZS )r
zCarrot compatible producer.� �directTFNc s� |r|}|p| j | _ |p | j| _|p(| j| _|d ur:|| _|d urH|| _t| j t�srt| j | j| j| j| jd�| _ t� j|| j fi |�� d S )N)�name�type� routing_key� auto_delete�durable) �exchange�
exchange_typer r r �
isinstancer �super�__init__) �selfr
r r r r r �channel�kwargs�� __class__r r r ) s" �zPublisher.__init__c O s | j |i |��S �N)�publish)r �argsr" r r r �send@ s zPublisher.sendc s t � �� d| _d S �NT)r �close�_closed�r r# r r r* C s
zPublisher.closec C s | S r% r r, r r r � __enter__G s zPublisher.__enter__�type[BaseException] | None�BaseException | None�TracebackType | None�None��exc_type�exc_val�exc_tb�returnc C s | � � d S r% �r* �r r3 r4 r5 r r r �__exit__J s zPublisher.__exit__c C s | j S r% )r! r, r r r �backendR s zPublisher.backend)NNNNNN)�__name__�
__module__� __qualname__�__doc__r r r r r r+ r r( r* r- r9 �propertyr: �
__classcell__r r r# r r
s" � r
c s� e Zd ZdZdZdZdZdZdZdZ dZ
dZ d$� fdd� Z � fd d
�Z
d d � Zd
d� Zddddd�dd�Zdd� Zd%dd�Zdd� Zd&d d �Zd'd d �Zd(d d!�Zd)d"d#�Z� ZS )*r zCarrot compatible consumer.r r TFNc
s� |� � | _|d ur|| _|d ur&|| _|d ur4|| _|p<| j| _|pH| j| _|pT| j| _|p`| j| _t | j| j| j| j| jd�}t
| j|| j| j| j| jd�}t � j | j|fi | �� d S )N)r r r r )r r r � exclusiver )
r! r: r rA r �queuer r r r r r r )
r r
rB r r r r rA r r" r# r r r c s2
��zConsumer.__init__c s || _ t� �|� d S r% �r: r �revive�r r! r# r r rD � s zConsumer.revivec C s | � � | j�� d| _d S r) )�cancelr: r* r+ r, r r r r* � s
zConsumer.closec C s | S r% r r, r r r r- � s zConsumer.__enter__r. r/ r0 r1 r2 c C s | � � d S r% r7 r8 r r r r9 � s zConsumer.__exit__c C s | j dd�S )NT)�infinite)� iterqueuer, r r r �__iter__� s zConsumer.__iter__c C s8 |d u r| j }| jd �|�}|r4|r4| �|j|� |S )Nr )r �queues�getZreceive�payload)r r Zenable_callbacks�messager r r �fetch� s zConsumer.fetchc C s t d��d S )Nz Use fetch(enable_callbacks=True))�NotImplementedErrorr, r r r � process_next� s zConsumer.process_nextc C s |d urt d��| �� S )Nz&discard_all does not implement filters)rO �purge)r �
filterfuncr r r � discard_all� s
�zConsumer.discard_allc C s t | j| ||�S r% �r r
�r r r r r r � iterconsume� s zConsumer.iterconsumec C s | � |�}t|�S r% )rV �list)r r �itr r r �wait� s
z
Consumer.waitc c s: t � D ].}| �� }|s |d u s*|r.||kr. q6|V qd S r% )r rN )r r rG Zitems_since_start�itemr r r rH � s
��zConsumer.iterqueue)NNNNNNN)NF)N)NN)N)NF)r; r< r= r> rB r r r r rA r r+ r rD r* r- r9 rI rN rP rS rV rY rH r@ r r r# r r W s0 �
r c sT e Zd Zd� fdd� Zddd�Zdd� Zd d
� Zd d � Z� fd
d�Zdd� Z � Z
S )� ConsumerSetNc
s� |rd| _ || _nd| _ |�� | _g }|r@|D ]}|�|j� q.|rn|�� D ] \}} |�tj|fi | ��� qLt � j
| j|fi |�� d S )NTF) �_provided_channelr: r! �extendrJ �items�appendr � from_dictr r )
r r
r` Z consumersr! r" rJ r Z
queue_nameZ
queue_optionsr# r r r � s
zConsumerSet.__init__Fc C s t | j| ||�S r% rT rU r r r rV � s zConsumerSet.iterconsumec C s | � � S r% )rQ r, r r r rS � s zConsumerSet.discard_allc K s | � tj|fi |���S r% )� add_queuer r` )r rB �optionsr r r �add_consumer_from_dict� s z"ConsumerSet.add_consumer_from_dictc C s |j D ]}| �|� qd S r% )rJ ra )r r rB r r r � add_consumer� s
zConsumerSet.add_consumerc s || _ t� �|� d S r% rC rE r# r r rD � s zConsumerSet.revivec C s | � � | js| j�� d S r% )rF r\ r! r* r, r r r r* � s zConsumerSet.close)NNN)NF) r; r< r= r rV rS rc rd rD r* r@ r r r# r r[ � s �
r[ )FN)r> �
__future__r � itertoolsr �typingr r r �entityr r �typesr �__all__r` Zentry_to_queuer ZProducerr
r r[ r r r r �<module> s
8c