Server : LiteSpeed System : Linux barito.iixcp.rumahweb.net 5.14.0-611.49.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Tue Apr 21 16:39:08 EDT 2026 x86_64 User : elvh3918 ( 1528) PHP Version : 8.2.31 Disable Function : mail Directory : /usr/local/lib/python3.9/site-packages/celery/backends/__pycache__/ |
a
��ShE( � @ s d Z ddlZddlZddlZddlmZ ddlmZ ddlmZ ddl m
Z
ddl m Z ddl
mZ dd lmZ dd
lmZ d Zi Zd d
� Zed�G dd� d��ZG dd� de�Zed�G dd� de��Zed�G dd� de��ZG dd� d�ZG dd � d �ZdS ) z$Async I/O backend support utilities.� N)�deque)�Empty)�sleep)�WeakKeyDictionary)�detect_environment)�states)� TimeoutError)�THREAD_TIMEOUT_MAX)�AsyncBackendMixin�BaseResultConsumer�Drainer�register_drainerc s � fdd�}|S )z5Decorator used to register a new result drainer type.c s | t � < | S �N)�drainers)�cls��name� �F/usr/local/lib/python3.9/site-packages/celery/backends/asynchronous.py�_inner s z register_drainer.<locals>._innerr )r r r r r r
s r
�defaultc @ s< e Zd ZdZdd� Zdd� Zdd� Zdd
d �Zdd d
�ZdS )r zResult draining service.c C s
|| _ d S r )�result_consumer)�selfr r r r �__init__$ s zDrainer.__init__c C s d S r r �r r r r �start' s z
Drainer.startc C s d S r r r r r r �stop* s z Drainer.stopN� c c st |p
| j j}t�� }|r0t�� | |kr0t�� �z| j|||d�V W n tjyZ Y n0 |rf|� |jrqpqd S �N��timeout)r � drain_events�time� monotonic�socketr �wait_for�ready)r �pr �interval� on_interval�waitZ
time_startr r r �drain_events_until- s zDrainer.drain_events_untilc C s ||d� d S r r �r r' r* r r r r r% > s zDrainer.wait_for)Nr NN)N) �__name__�
__module__� __qualname__�__doc__r r r r+ r% r r r r r s
r c sZ e Zd ZdZdZdZdd� Zdd� Z� fdd�Zdd � Z d
d � Z
d d
� Z ddd�Z � Z
S )�greenletDrainerNc C s dS )z,create new self._drain_complete_event objectNr r r r r � _create_drain_complete_eventG s z,greenletDrainer._create_drain_complete_eventc C s dS )z5raise self._drain_complete_event for wakeup .wait_forNr r r r r �_send_drain_complete_eventK s z*greenletDrainer._send_drain_complete_eventc s< t � j|i |�� t�� | _t�� | _t�� | _| �� d S r )�superr � threading�Event�_started�_stopped� _shutdownr2 )r �args�kwargs�� __class__r r r O s
zgreenletDrainer.__init__c C s\ | j �� | j�� sNz"| jjdd� | �� | �� W q
tj yJ Y q
0 q
| j
�� d S )Nr r ) r7 �setr8 �is_setr r! r3 r2 r$ r r9 r r r r �runV s
zgreenletDrainer.runc C s& | j �� s"| �| j�| _| j �� d S r )r7 r? �spawnr@ �_gr* r r r r r a s
zgreenletDrainer.startc C s" | j �� | �� | j�t� d S r )r8 r> r3 r9 r* r r r r r r f s
zgreenletDrainer.stopc C s | � � |js | jj|d� d S r )r r& �_drain_complete_eventr* r, r r r r% k s zgreenletDrainer.wait_for)N)r- r. r/ rA rB rC r2 r3 r r@ r r r% �
__classcell__r r r< r r1 B s r1 �eventletc @ s$ e Zd Zdd� Zdd� Zdd� ZdS )�eventletDrainerc C s$ ddl m}m} ||�}|d� |S )Nr )r rA )rE r rA )r �funcr rA �gr r r rA t s zeventletDrainer.spawnc C s ddl m} |� | _d S �Nr )r6 )Zeventlet.eventr6 rC �r r6 r r r r2 z s z,eventletDrainer._create_drain_complete_eventc C s | j �� d S r )rC �sendr r r r r3 ~ s z*eventletDrainer._send_drain_complete_eventN�r- r. r/ rA r2 r3 r r r r rF q s rF �geventc @ s$ e Zd Zdd� Zdd� Zdd� ZdS )�
geventDrainerc C s dd l }|�|�}|�d� |S )Nr )rM rA r )r rG rM rH r r r rA � s
zgeventDrainer.spawnc C s ddl m} |� | _d S rI )Z gevent.eventr6 rC rJ r r r r2 � s z*geventDrainer._create_drain_complete_eventc C s | j �� | �� d S r )rC r> r2 r r r r r3 � s
z(geventDrainer._send_drain_complete_eventNrL r r r r rN � s rN c @ s� e Zd ZdZdd� Zd dd�Zd dd �Zd
d � Zd d d
�Zd dd�Z dd� Z
dd� Z dd� Z d!dd�Z
d"dd�Zedd � �ZdS )#r
z.Mixin for backends that enables the async API.c C s || j j|< d S r )r �buckets)r �result�bucketr r r �
_collect_into� s z AsyncBackendMixin._collect_intoTc k s� | � � |j}|st� �t� }|D ]8}t|d�s<|�|� q"|jrN|�|� q"| �||� q"| j|fd|i|��D ]:}|rr|� � }t|d�s�|j
|j fV qv|j
|jfV qvqr|r�|� � }|j
|jfV q�d S )N�_cache�no_ack) �_ensure_not_eager�results�
StopIterationr �hasattr�appendrS rR �_wait_for_pending�popleft�id�children)r rP rT r; rV rQ �node�_r r r � iter_native� s(
z AsyncBackendMixin.iter_nativeFc C sH |r| j j�� z| �|� W n$ tyB | j|j||d� Y n0 |S )N)�weak)r �drainerr �_maybe_resolve_from_bufferr �_add_pending_resultr\ )r rP ra �
start_drainerr r r �add_pending_result� s z$AsyncBackendMixin.add_pending_resultc C s |� | j�|j�� d S r )�_maybe_set_cache�_pending_messagesZtaker\ �r rP r r r rc � s z,AsyncBackendMixin._maybe_resolve_from_bufferc C s< | j \}}||vr8|j|vr8||r&|n||< | j�|� d S r )�_pending_resultsr\ r � consume_from)r �task_idrP ra ZconcreteZweak_r r r rd � s
z%AsyncBackendMixin._add_pending_resultc s � j j�� � �fdd�|D �S )Nc s g | ]}� j |�d d��qS )F)ra re )rf )�.0rP �r ra r r �
<listcomp>� s �z9AsyncBackendMixin.add_pending_results.<locals>.<listcomp>)r rb r )r rV ra r rn r �add_pending_results� s �z%AsyncBackendMixin.add_pending_resultsc C s | � |j� | �|� |S r )�_remove_pending_resultr\ �on_result_fulfilledri r r r �remove_pending_result� s
z'AsyncBackendMixin.remove_pending_resultc C s | j D ]}|�|d � qd S r )rj �pop�r rl �mappingr r r rq � s
z(AsyncBackendMixin._remove_pending_resultc C s | j �|j� d S r )r �
cancel_forr\ ri r r r rr � s z%AsyncBackendMixin.on_result_fulfilledNc K s. | � � | j|fi |��D ]}q|j||d�S )N)�callback� propagate)rU rZ Z maybe_throw)r rP rx ry r; r_ r r r �wait_for_pending� s z"AsyncBackendMixin.wait_for_pendingc K s | j j|f|||d�|��S )N)r r) �
on_message)r rZ )r rP r r) r{ r; r r r rZ � s ���z#AsyncBackendMixin._wait_for_pendingc C s dS )NTr r r r r �is_async� s zAsyncBackendMixin.is_async)T)FT)F)F)NT)NNN)r- r. r/ r0 rR r` rf rc rd rp rs rq rr rz rZ �propertyr| r r r r r
� s
�
�
r
c @ s� e Zd ZdZdd� Zdd� Zdd� Zd d d
�Zd d � Zd
d� Z dd� Z
dd� Z d dd�Z d!dd�Z
d"dd�Zdd� Zdd � Zd d � ZdS )#r z2Manager responsible for consuming result messages.c C s@ || _ || _|| _|| _|| _d | _t� | _tt � | �| _
d S r ) �backend�app�acceptrj rh r{ r rO r r rb )r r~ r r� Zpending_resultsZpending_messagesr r r r � s zBaseResultConsumer.__init__c K s
t � �d S r ��NotImplementedError)r Zinitial_task_idr; r r r r � s zBaseResultConsumer.startc C s d S r r r r r r r s zBaseResultConsumer.stopNc C s
t � �d S r r� )r r r r r r! s z BaseResultConsumer.drain_eventsc C s
t � �d S r r� �r rl r r r rk s z BaseResultConsumer.consume_fromc C s
t � �d S r r� r� r r r rw s z BaseResultConsumer.cancel_forc C s$ | j �� t� | _ d | _| �� d S r )rO �clearr r{ �
on_after_forkr r r r � _after_fork s
z BaseResultConsumer._after_forkc C s d S r r r r r r r� s z BaseResultConsumer.on_after_forkc C s | j j|||d�S )N�r r) )rb r+ )r r' r r) r r r r+ s �z%BaseResultConsumer.drain_events_untilc k s� | j |fd|i|�� | j| }| _zRz*| j|j||d�D ]}d V td� q:W n tjyl td��Y n0 W || _n|| _0 d S )Nr r� r zThe operation timed out.)�on_wait_for_pendingr{ r+ Zon_readyr r$ r r )r rP r r) r{ r; Z prev_on_mr_ r r r rZ s �
z$BaseResultConsumer._wait_for_pendingc K s d S r r )r rP r r; r r r r� ) s z&BaseResultConsumer.on_wait_for_pendingc C s | � |j|� d S r )�on_state_change�payload)r �messager r r �on_out_of_band_result, s z(BaseResultConsumer.on_out_of_band_resultc C s: | j D ]&}z|| W S ty* Y q0 qt|��d S r )rj �KeyErrorru r r r �_get_pending_result/ s
z&BaseResultConsumer._get_pending_resultc C s� | j r| � |� |d tjv r�|d }z| �|�}W n tyT | j�||� Y n>0 |�|� | j}z|� |�}W n ty� Y n 0 |�
|� t d� d S )N�statusrl r ) r{ r Z READY_STATESr� r� rh �putrg rO rt rY r )r �metar� rl rP rO rQ r r r r� 7 s
z"BaseResultConsumer.on_state_change)N)NN)NNN)N)r- r. r/ r0 r r r r! rk rw r� r� r+ rZ r� r� r� r� r r r r r � s
�
r ) r0 r$ r5 r" � collectionsr �queuer r �weakrefr Zkombu.utils.compatr Zceleryr Zcelery.exceptionsr Zcelery.utils.threadsr �__all__r r
r r1 rF rN r
r r r r r �<module> s. !/[