File: //home/arjun/projects/env/lib64/python3.10/site-packages/kombu/__pycache__/compat.cpython-310.pyc
o
'we` � @ s� d Z ddlmZ ddlmZ ddlmZ ddlmZ ddl m
Z
m Z er*ddl 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 s< � |j |d� td�D ]}|r||kr d S | �� V q d S )N)�no_ackr )�consumer � drain_events)�
connection�consumerr �limit� iteration� r �E/home/arjun/projects/env/lib/python3.10/site-packages/kombu/compat.py� _iterconsume s � �r c sr e Zd ZdZdZdZdZdZdZdZ d� fdd� Z
d d
� Z � fd d �Z d
d� Z
d dd�Zedd� �Z� ZS ) r
zCarrot compatible producer.� �directTFNc s� |r|}|p| j | _ |p| j| _|p| j| _|d ur || _|d ur$|| _t| j t�s9t| 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@ � zPublisher.sendc s t � �� d| _d S �NT)r �close�_closed�r" r% r r r- C s
zPublisher.closec C � | S r' r r/ r r r � __enter__G � zPublisher.__enter__�exc_type�type[BaseException] | None�exc_val�BaseException | None�exc_tb�TracebackType | None�return�Nonec C � | � � d S r' �r- �r" r3 r5 r7 r r r �__exit__J � zPublisher.__exit__c C s | j S r' )r# r/ r r r �backendR s zPublisher.backend)NNNNNN�r3 r4 r5 r6 r7 r8 r9 r: )�__name__�
__module__� __qualname__�__doc__r r r r r r. r! r* r- r1 r> �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� Zd(dd�Zdd� Zd)dd �Zd d � 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 ur|| _|p | j| _|p$| j| _|p*| j| _|p0| 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 rH r �queuer r r r r r r! )
r" r rI r r r r rH r r$ r% r r r! c s2
�� zConsumer.__init__c � || _ t� �|� d S r' �r@ r �revive�r" r# r% r r rL � � zConsumer.revivec C s | � � | j�� d| _d S r, )�cancelr@ r- r. r/ r r r r- � s
zConsumer.closec C r0 r' r r/ r r r r1 � r2 zConsumer.__enter__r3 r4 r5 r6 r7 r8 r9 r: c C r; r' r< r= r r r r> � r? 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 �|�}|r|r| �|j|� |S )Nr )r �queues�get�receive�payload)r" r �enable_callbacks�messager r r �fetch� s zConsumer.fetchc C s t d��)Nz Use fetch(enable_callbacks=True))�NotImplementedErrorr/ r r r � process_next� � zConsumer.process_nextc C s |d urt d��| �� S )Nz&discard_all does not implement filters)rZ �purge)r" �
filterfuncr r r � discard_all� s
�zConsumer.discard_allc C � t | j| ||�S r' �r r �r" r r r r r � iterconsume� r+ zConsumer.iterconsumec C s | � |�}t|�S r' )rc �list)r" r �itr r r �wait� s
z
Consumer.waitc c s> � t � D ]}| �� }|s|d u s|r||kr d S |V qd S r' )r rY )r" r rP �items_since_start�itemr r r rQ � s �
��zConsumer.iterqueue)NNNNNNNrA �NFr' )NN)rB rC rD rE rI r r r r rH r r. r! rL r- r1 r> rR rY r[ r_ rc rf rQ rG r r r% r r W s2 �
r c sX e Zd Z d� fdd� Zddd�Zdd� Zd d
� Zd d � Z� fd
d�Zdd� Z � Z
S )� ConsumerSetNc
s� |r d| _ || _nd| _ |�� | _g }|r |D ]}|�|j� q|r7|�� D ]\}} |�tj|fi | ��� q&t � j
| j|fi |�� d S )NTF) �_provided_channelr@ r# �extendrS �items�appendr � from_dictr r! )
r" r ro � consumersr# r$ rS r �
queue_name�
queue_optionsr% r r r! � s
zConsumerSet.__init__Fc C r` r' ra rb r r r rc � r+ zConsumerSet.iterconsumec C s | � � S r' )r] r/ r r r r_ � r\ zConsumerSet.discard_allc K s | � tj|fi |���S r' )� add_queuer ro )r" rI �optionsr r r �add_consumer_from_dict� s z"ConsumerSet.add_consumer_from_dictc C s |j D ]}| �|� qd S r' )rS rs )r" r rI r r r � add_consumer� s
�zConsumerSet.add_consumerc rJ r' rK rM r% r r rL � rN zConsumerSet.revivec C s | � � | js| j�� d S d S r' )rO rk r# r- r/ r r r r- � s �zConsumerSet.close)NNNri ) rB rC rD r! rc r_ ru rv rL r- rG r r r% r rj � s �
rj )FN)rE �
__future__r � itertoolsr �typingr r r �entityr r �typesr �__all__ro �entry_to_queuer �Producerr
r rj r r r r �<module> s
8c