File: //usr/local/lib/python3.9/site-packages/celery/events/__pycache__/receiver.cpython-39.pyc
a
X>h� � @ s� d Z ddlZddlmZ ddlmZ ddlmZ ddlm Z ddl
m Z ddl m
Z
dd lmZ d
d lmZ d Zd
Zedd�ZG dd� de �ZdS )z Event receiver implementation.� N)�
itemgetter)�Queue)�
maybe_channel)�
ConsumerMixin)�uuid)�app_or_default)�adjust_timestamp� )� get_exchange)�
EventReceiver���� utcoffset� timestampc @ s� e Zd ZdZdZddd�Zdd� Zdd � Zdd d �Zdd
d�Z d dd�Z
d dd�Z d
e j e
eefdd�Zeefdd�Zedd� �ZdS ) r a? Capture events.
Arguments:
connection (kombu.Connection): Connection to the broker.
handlers (Mapping[Callable]): Event handlers.
This is a map of event type names and their handlers.
The special handler `"*"` captures all events that don't have a
handler.
N�#c
C s� t |p
| j�| _t|�| _|d u r&i n|| _|| _|p:t� | _|pJ| jjj | _
t | j p^| j�
� | jjjd�| _|d u r�| jjj}| d u r�| jjj} td�| j
| jg�| j| jdd|| d�| _| jj| _| jj| _| jj| _|d u r�| jjjdh}|| _d S )N)�name�.TF)�exchange� routing_keyZ auto_deleteZdurableZ message_ttl�expires�json) r �appr �channel�handlersr r �node_id�confZevent_queue_prefix� queue_prefixr
�
connectionZconnection_for_writeZevent_exchanger Zevent_queue_ttlZevent_queue_expiresr �join�queue�clock�adjust� adjust_clockZforward�
forward_clockZevent_serializer�accept)
�selfr r r r r r r# Z queue_ttlZ
queue_expires� r% �@/usr/local/lib/python3.9/site-packages/celery/events/receiver.py�__init__# s8
�
�
zEventReceiver.__init__c C s( | j �|�p| j �d�}|o"||� dS )z3Process event by dispatching to configured handler.�*N)r �get)r$ �type�event�handlerr% r% r& �processB s zEventReceiver.processc C s || j g| jgd| jd�gS )NT)�queues� callbacksZno_ackr# )r �_receiver# )r$ ZConsumerr r% r% r& �
get_consumersG s �zEventReceiver.get_consumersTc K s |r| j |d� d S )N)r )�wakeup_workers)r$ r r Z consumers�wakeup�kwargsr% r% r& �on_consume_readyL s z EventReceiver.on_consume_readyc C s | j |||d�S )N��limit�timeoutr3 �Zconsume)r$ r7 r8 r3 r% r% r& � itercaptureQ s zEventReceiver.itercapturec C s | j |||d�D ]}qdS )z�Open up a consumer capturing events.
This has to run in the main process, and it will never stop
unless :attr:`EventDispatcher.should_stop` is set to True, or
forced via :exc:`KeyboardInterrupt` or :exc:`SystemExit`.
r6 Nr9 )r$ r7 r8 r3 �_r% r% r& �captureT s zEventReceiver.capturec C s | j jjd| j|d� d S )NZ heartbeat)r r )r �control� broadcastr )r$ r r% r% r& r2 ^ s
�z EventReceiver.wakeup_workersc C s� |d }|dkr4| j jpd| }|d< | �|� n6z |d } W n ty^ | �� |d< Y n 0 | �| � |r�z||�\}
} W n ty� Y n0 || |
�|d< |� |d< ||fS )Nr* z task-sentr r r Zlocal_received)r �valuer! �KeyErrorr" ) r$ �bodyZlocalize�nowZtzfieldsr �CLIENT_CLOCK_SKEWr* Z_cr �offsetr r% r% r& �event_from_messagec s"
z EventReceiver.event_from_messagec sB |||�r.| j | j �� � �fdd�|D � n| j | �|�� d S )Nc s g | ]}�� |�� �qS r% r% )�.0r+ �Z from_messager- r% r& �
<listcomp>� � z*EventReceiver._receive.<locals>.<listcomp>)r- rE )r$ rA �message�list�
isinstancer% rG r& r0 ~ s
zEventReceiver._receivec C s | j r| j jjS d S )N)r r �client)r$ r% r% r& r � s zEventReceiver.connection)Nr NNNNNN)T)NNT)NNT)N)�__name__�
__module__� __qualname__�__doc__r r' r- r1 r5 r: r< r2 �time� _TZGETTERr rC rE rK rL r0 �propertyr r% r% r% r& r s*
�
�
�
r )rQ rR �operatorr Zkombur Zkombu.connectionr Z kombu.mixinsr Zceleryr Z
celery.appr Zcelery.utils.timer r+ r
�__all__rC rS r r% r% r% r& �<module> s