File: //usr/local/lib/python3.9/site-packages/celery/events/__pycache__/dispatcher.cpython-39.pyc
a
X>h# � @ s� d Z ddlZddlZddlZddlmZmZ ddlmZ ddl m
Z
ddl m Z ddl
mZ dd lmZmZmZ d
ZG d d � d �ZdS )
z Event dispatcher sends events.� N)� defaultdict�deque)�Producer)�app_or_default)�
anon_nodename)� utcoffset� )�Event� get_exchange�
group_from)�EventDispatcherc @ s� e Zd ZdZdhZdZdZdZd"dd�Zd d
� Z d d � Z
d
d� Z dd� Z de
fdd�Zddefdd�Zdedde
fdd�Zd#dd�Zdd� Zd d � Zd d � Zd d!� Zeee�ZdS )$r a0 Dispatches event messages.
Arguments:
connection (kombu.Connection): Connection to the broker.
hostname (str): Hostname to identify ourselves as,
by default uses the hostname returned by
:func:`~celery.utils.anon_nodename`.
groups (Sequence[str]): List of groups to send events for.
:meth:`send` will ignore send requests to groups not in this list.
If this is :const:`None`, all events will be sent.
Example groups include ``"task"`` and ``"worker"``.
enabled (bool): Set to :const:`False` to not actually publish any
events, making :meth:`send` a no-op.
channel (kombu.Channel): Can be used instead of `connection` to specify
an exact channel to use when sending events.
buffer_while_offline (bool): If enabled events will be buffered
while the connection is down. :meth:`flush` must be called
as soon as the connection is re-established.
Note:
You need to :meth:`close` this after use.
ZsqlNTr � c
C s4 t |p
| j�| _|| _|| _|p$t� | _|| _|
p6t� | _| | _ | | _
t t �| _
t�� | _d | _t� | _|pt| jjj| _t� | _t� | _t|p�g �| _tj tj g| _ | jj | _ | | _ |s�|r�|jj!| _|| _"| jp�| j�#� }
t$|
| jjj%d�| _&|
j'j(| j)v �r
d| _"| j"�r| �*� d| ji| _+t,�-� | _.d S )N)�nameF�hostname)/r �app�
connection�channelr r �buffer_while_offline� frozenset� buffer_group� buffer_limit�on_send_bufferedr �list�
_group_buffer� threading�Lock�mutex�producerr �_outbound_buffer�confZevent_serializer�
serializer�set�
on_enabled� on_disabled�groups�time�timezone�altzone�tzoffset�clock�
delivery_mode�client�enabledZconnection_for_writer
Zevent_exchange�exchange� transportZ driver_type�DISABLED_TRANSPORTS�enable�headers�os�getpid�pid)�selfr r r, r r r r r$ r* r r r Zconninfo� r6 �B/usr/local/lib/python3.9/site-packages/celery/events/dispatcher.py�__init__: s@
� zEventDispatcher.__init__c C s | S �Nr6 �r5 r6 r6 r7 � __enter__^ s zEventDispatcher.__enter__c G s | � � d S r9 )�close)r5 �exc_infor6 r6 r7 �__exit__a s zEventDispatcher.__exit__c C s: t | jp | j| j| jdd�| _d| _| jD ]
}|� q*d S )NF)r- r Z auto_declareT)r r r r- r r r, r" �r5 �callbackr6 r6 r7 r0 d s �
zEventDispatcher.enablec C s* | j r&d| _ | �� | jD ]
}|� qd S )NF)r, r< r# r? r6 r6 r7 �disablem s
zEventDispatcher.disableFc K s| |rdn| j �� }||f| jt� | j|d�|��}| j�0 | j||fd|�dd�i|��W d � S 1 sn0 Y dS )au Publish event using custom :class:`~kombu.Producer`.
Arguments:
type (str): Event type name, with group separated by dash (`-`).
fields: Dictionary of event fields, must be json serializable.
producer (kombu.Producer): Producer instance to use:
only the ``publish`` method will be called.
retry (bool): Retry in the event of connection failure.
retry_policy (Mapping): Map of custom retry policy options.
See :meth:`~kombu.Connection.ensure`.
blind (bool): Don't set logical clock value (also don't forward
the internal logical clock).
Event (Callable): Event type used to create event.
Defaults to :func:`Event`.
utcoffset (Callable): Function returning the current
utc offset in hours.
N�r r r4 r) � routing_key�-�.)r) �forwardr r r4 r �_publish�replace) r5 �type�fieldsr �blindr �kwargsr) �eventr6 r6 r7 �publisht s ��
��zEventDispatcher.publishc C st | j }z*|j|||j|||g| j| j| jd� W n> tyn } z&| jsH� | j� |||f� W Y d }~n
d }~0 0 d S )N)rC r- �retry� retry_policyZdeclarer r1 r* )
r- rN r r r1 r* � Exceptionr r �append) r5 rM r rC rO rP r r- �excr6 r6 r7 rG � s" �
zEventDispatcher._publishc
K s� | j r�| jt|� }} |r&| |vr&dS | | jv r�| j�� }
||f| j|� | j|
d�|��} | j| } | � | � t
| �| j kr�| � � q�| j
r�| �
� n| j||| j||||d�S dS )a� Send event.
Arguments:
type (str): Event type name, with group separated by dash (`-`).
retry (bool): Retry in the event of connection failure.
retry_policy (Mapping): Map of custom retry policy options.
See :meth:`~kombu.Connection.ensure`.
blind (bool): Don't set logical clock value (also don't forward
the internal logical clock).
Event (Callable): Event type used to create event,
defaults to :func:`Event`.
utcoffset (Callable): unction returning the current utc offset
in hours.
**fields (Any): Event fields -- must be json serializable.
NrB )rK r rO rP )r, r$ r r r) rF r r4 r rR �lenr �flushr rN r )
r5 rI rK r rO rP r rJ r$ �groupr) rM �bufr6 r6 r7 �send� s,
��
�zEventDispatcher.sendc C s� |rpt | j�}zT| j�0 |D ]\}}}| �|| j|� q W d � n1 sL0 Y W | j�� n | j�� 0 |r�| j�D | j�� D ](\}}| �|| jd| � g |dd�< q�W d � n1 s�0 Y dS )zFlush the outbound buffer.Nz%s.multi)r r r rG r �clearr �items) r5 �errorsr$ rW rM rC �_rV �eventsr6 r6 r7 rU � s
2zEventDispatcher.flushc C s | j �|j � dS )z-Copy the outbound buffer of another instance.N)r �extend)r5 �otherr6 r6 r7 �
extend_buffer� s z EventDispatcher.extend_bufferc C s | j �� o| j �� d| _dS )zClose the event dispatcher.N)r �locked�releaser r: r6 r6 r7 r< � s zEventDispatcher.closec C s | j S r9 �r r: r6 r6 r7 �_get_publisher� s z EventDispatcher._get_publisherc C s
|| _ d S r9 rc )r5 r r6 r6 r7 �_set_publisher� s z EventDispatcher._set_publisher) NNTNTNNNr Nr
N)TT)�__name__�
__module__� __qualname__�__doc__r/ r r" r# r8 r; r> r0 rA r rN r rG rX rU r` r< rd re �propertyZ publisherr6 r6 r6 r7 r s8 �
$ �
�
�
%
r )ri r2 r r% � collectionsr r Zkombur Z
celery.appr Zcelery.utils.nodenamesr Zcelery.utils.timer rM r r
r �__all__r r6 r6 r6 r7 �<module> s