o
    wvXj^%  ã                   @   s  d Z ddlmZmZ ddlZddlZddlmZ ddlm	Z	 ddl
mZ ddlmZ ddlmZ dd	lmZ dd
lmZmZ ddlmZ dZi Zdd„ ZedƒG dd„ deƒƒZG dd„ deƒZedƒG dd„ deƒƒZedƒG dd„ deƒƒZG dd„ deƒZG dd„ deƒZ dS )z$Async I/O backend support utilities.é    )Úabsolute_importÚunicode_literalsN)Údeque)Úsleep)ÚWeakKeyDictionary)Údetect_environment)Ústates)ÚTimeoutError)ÚEmptyÚ	monotonic)ÚTHREAD_TIMEOUT_MAX)ÚAsyncBackendMixinÚBaseResultConsumerÚDrainerÚregister_drainerc                    s   ‡ fdd„}|S )z5Decorator used to register a new result drainer type.c                    s   | t ˆ < | S ©N)Údrainers)Úcls©Úname© úY/var/www/html/myproject/venv/lib/python3.10/site-packages/celery/backends/asynchronous.pyÚ_inner   s   z register_drainer.<locals>._innerr   )r   r   r   r   r   r      s   r   Údefaultc                   @   s<   e Zd ZdZdd„ Zdd„ Zdd„ Zdd
d„Zddd„ZdS )r   zResult draining service.c                 C   s
   || _ d S r   )Úresult_consumer)Úselfr   r   r   r   Ú__init__%   s   
zDrainer.__init__c                 C   ó   d S r   r   ©r   r   r   r   Ústart(   ó   zDrainer.startc                 C   r   r   r   r   r   r   r   Ústop+   r    zDrainer.stopNé   c                 c   sr   � |p| j j}tƒ }	 |rtƒ | |krt ¡ ‚z| j|||d�V  W n
 tjy-   Y nw |r3|ƒ  |jr8d S q©Nr"   ©Útimeout)r   Údrain_eventsr   Úsocketr%   Úwait_forÚready)r   Úpr%   ÚintervalÚon_intervalÚwaitÚ
time_startr   r   r   Údrain_events_until.   s    €ÿõzDrainer.drain_events_untilc                 C   s   ||d� d S ©Nr$   r   ©r   r*   r-   r%   r   r   r   r(   ?   s   zDrainer.wait_for)Nr"   NNr   )	Ú__name__Ú
__module__Ú__qualname__Ú__doc__r   r   r!   r/   r(   r   r   r   r   r   !   s    
r   c                       s<   e Zd ZdZdZ‡ fdd„Zdd„ Zdd„ Zdd	„ Z‡  Z	S )
ÚgreenletDrainerNc                    s8   t t| ƒj|i |¤Ž t ¡ | _t ¡ | _t ¡ | _d S r   )Úsuperr6   r   Ú	threadingÚEventÚ_startedÚ_stoppedÚ	_shutdown)r   ÚargsÚkwargs©Ú	__class__r   r   r   G   s   

zgreenletDrainer.__init__c                 C   sT   | j  ¡  | j ¡ s#z	| jjdd� W n
 tjy   Y nw | j ¡ r
| j ¡  d S r#   )	r:   Úsetr;   Úis_setr   r&   r'   r%   r<   r   r   r   r   ÚrunM   s   

ÿ
ýzgreenletDrainer.runc                 C   s*   | j  ¡ s|  | j¡| _| j  ¡  d S d S r   )r:   rB   ÚspawnrC   Ú_gr-   r   r   r   r   r   V   s   
þzgreenletDrainer.startc                 C   s   | j  ¡  | j t¡ d S r   )r;   rA   r<   r-   r   r   r   r   r   r!   [   s   
zgreenletDrainer.stop)
r2   r3   r4   rD   rE   r   rC   r   r!   Ú__classcell__r   r   r?   r   r6   C   s    	r6   Úeventletc                   @   ó   e Zd Zdd„ Zddd„ZdS )ÚeventletDrainerc                 C   ó$   ddl m}m} ||ƒ}|dƒ |S ©Nr   )rD   r   )rG   rD   r   ©r   ÚfuncrD   r   Úgr   r   r   rD   c   ó   zeventletDrainer.spawnNc                 C   s&   |   ¡  |js| jjj|d� d S d S r0   )r   r)   rE   Ú_exit_eventr-   r1   r   r   r   r(   i   s   ÿzeventletDrainer.wait_forr   ©r2   r3   r4   rD   r(   r   r   r   r   rI   `   ó    rI   Úgeventc                   @   rH   )ÚgeventDrainerc                 C   rJ   rK   )rS   rD   r   rL   r   r   r   rD   r   rO   zgeventDrainer.spawnNc                 C   s0   dd l }|  ¡  |js|j| jg|d� d S d S )Nr   r$   )rS   r   r)   r-   rE   )r   r*   r-   r%   rS   r   r   r   r(   x   s
   ÿzgeventDrainer.wait_forr   rQ   r   r   r   r   rT   o   rR   rT   c                   @   s„   e Zd ZdZdd„ Zddd„Zddd	„Zd
d„ Zddd„Zddd„Z	dd„ Z
dd„ Zdd„ Z	d dd„Z	d!dd„Zedd„ ƒZdS )"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}|stƒ ‚tƒ }|D ]}t|dƒs| |¡ q|jr(| |¡ q|  ||¡ q| j|fd|i|¤ŽD ]}|rX| 	¡ }t|dƒsO|j
|jfV  n|j
|jfV  |s>q:|rj| 	¡ }|j
|jfV  |s[d S d S )NÚ_cacheÚno_ack)Ú_ensure_not_eagerÚresultsÚStopIterationr   ÚhasattrÚappendrY   rX   Ú_wait_for_pendingÚpopleftÚidÚchildren)r   rV   rZ   r>   r\   rW   ÚnodeÚ_r   r   r   Úiter_native…   s0   €

û€þzAsyncBackendMixin.iter_nativeFc                 C   sH   |r| j j ¡  z|  |¡ W |S  ty#   | j|j||d� Y |S w )N)Úweak)r   Údrainerr   Ú_maybe_resolve_from_bufferr
   Ú_add_pending_resultrb   )r   rV   rg   Ústart_drainerr   r   r   Úadd_pending_result¢   s   þþz$AsyncBackendMixin.add_pending_resultc                 C   s   |  | j |j¡¡ d S r   )Ú_maybe_set_cacheÚ_pending_messagesÚtakerb   ©r   rV   r   r   r   ri   «   s   z,AsyncBackendMixin._maybe_resolve_from_bufferc                 C   sD   | j \}}||vr|j|vr ||r|n||< | j |¡ d S d S d S r   )Ú_pending_resultsrb   r   Úconsume_from)r   Útask_idrV   rg   ÚconcreteÚweak_r   r   r   rj   ®   s
   
þz%AsyncBackendMixin._add_pending_resultc                    s    ˆ j j ¡  ‡ ‡fdd„|D ƒS )Nc                    s   g | ]
}ˆ j |ˆd d�‘qS )F)rg   rk   )rl   )Ú.0rV   ©r   rg   r   r   Ú
<listcomp>¶   s    ÿz9AsyncBackendMixin.add_pending_results.<locals>.<listcomp>)r   rh   r   )r   r\   rg   r   rw   r   Úadd_pending_results´   s   ÿz%AsyncBackendMixin.add_pending_resultsc                 C   s   |   |j¡ |  |¡ |S r   )Ú_remove_pending_resultrb   Úon_result_fulfilledrp   r   r   r   Úremove_pending_result¹   s   
z'AsyncBackendMixin.remove_pending_resultc                 C   s   | j D ]}| |d ¡ qd S r   )rq   Úpop©r   rs   Úmappingr   r   r   rz   ¾   s   
ÿz(AsyncBackendMixin._remove_pending_resultc                 C   s   | j  |j¡ d S r   )r   Ú
cancel_forrb   rp   r   r   r   r{   Â   ó   z%AsyncBackendMixin.on_result_fulfilledNc                 K   s.   |   ¡  | j|fi |¤ŽD ]}q|j||d�S )N)ÚcallbackÚ	propagate)r[   r`   Úmaybe_throw)r   rV   r‚   rƒ   r>   re   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   r`   )r   rV   r%   r,   r†   r>   r   r   r   r`   Ì   s   ÿþýz#AsyncBackendMixin._wait_for_pendingc                 C   s   dS ©NTr   r   r   r   r   Úis_asyncÕ   s   zAsyncBackendMixin.is_async)T)FT)Fr‡   ©NNN)r2   r3   r4   r5   rX   rf   rl   ri   rj   ry   r|   rz   r{   r…   r`   Úpropertyrˆ   r   r   r   r   r      s"    

	


ÿ
ÿ	r   c                   @   sŠ   e Zd ZdZdd„ Zdd„ Zdd„ Zdd	d
„Zdd„ Zdd„ Z	dd„ Z
dd„ Zd dd„Z	d!dd„Zddd„Zdd„ Zdd„ Zdd„ ZdS )"r   z2Manager responsible for consuming result messages.c                 C   s@   || _ || _|| _|| _|| _d | _tƒ | _tt	ƒ  | ƒ| _
d S r   )ÚbackendÚappÚacceptrq   rn   r†   r   rU   r   r   rh   )r   r‹   rŒ   r�   Úpending_resultsÚpending_messagesr   r   r   r   Ý   s   zBaseResultConsumer.__init__c                 K   ó   t ƒ ‚r   ©ÚNotImplementedError)r   Úinitial_task_idr>   r   r   r   r   è   ó   zBaseResultConsumer.startc                 C   r   r   r   r   r   r   r   r!   ë   r    zBaseResultConsumer.stopNc                 C   r�   r   r‘   )r   r%   r   r   r   r&   î   r”   zBaseResultConsumer.drain_eventsc                 C   r�   r   r‘   ©r   rs   r   r   r   rr   ñ   r”   zBaseResultConsumer.consume_fromc                 C   r�   r   r‘   r•   r   r   r   r€   ô   r”   zBaseResultConsumer.cancel_forc                 C   s$   | j  ¡  tƒ | _ d | _|  ¡  d S r   )rU   Úclearr   r†   Úon_after_forkr   r   r   r   Ú_after_fork÷   s   
zBaseResultConsumer._after_forkc                 C   r   r   r   r   r   r   r   r—   ý   r    z BaseResultConsumer.on_after_forkc                 C   s   | j j|||d�S )N©r%   r,   )rh   r/   )r   r*   r%   r,   r   r   r   r/      s   ÿz%BaseResultConsumer.drain_events_untilc                 k   s€   � | j |fd|i|¤Ž | j|}| _z(z| j|j||d�D ]	}d V  tdƒ qW n tjy5   tdƒ‚w W || _d S || _w )Nr%   r™   r   zThe operation timed out.)Úon_wait_for_pendingr†   r/   Úon_readyr   r'   r%   r	   )r   rV   r%   r,   r†   r>   Ú	prev_on_mre   r   r   r   r`     s    €
þ
üÿûz$BaseResultConsumer._wait_for_pendingc                 K   r   r   r   )r   rV   r%   r>   r   r   r   rš     r    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  r�   z(BaseResultConsumer.on_out_of_band_resultc              	   C   s4   | j D ]}z|| W   S  ty   Y qw t|ƒ‚r   )rq   ÚKeyErrorr~   r   r   r   Ú_get_pending_result  s   
ÿz&BaseResultConsumer._get_pending_resultc                 C   sž   | j r|   |¡ |d tjv rI|d }z|  |¡}W n ty*   | j ||¡ Y nw | |¡ | j}z| 	|¡}W n	 tyC   Y nw | 
|¡ tdƒ d S )NÚstatusrs   r   )r†   r   ÚREADY_STATESr¢   r¡   rn   Úputrm   rU   r}   r_   r   )r   ÚmetarŸ   rs   rV   rU   rW   r   r   r   r�   "  s$   
ý
ÿ
z"BaseResultConsumer.on_state_changer   )NNr‰   )r2   r3   r4   r5   r   r   r!   r&   rr   r€   r˜   r—   r/   r`   rš   r    r¢   r�   r   r   r   r   r   Ú   s"    


ÿ
r   )!r5   Ú
__future__r   r   r'   r8   Úcollectionsr   Útimer   Úweakrefr   Úkombu.utils.compatr   Úceleryr   Úcelery.exceptionsr	   Úcelery.fiver
   r   Úcelery.utils.threadsr   Ú__all__r   r   Úobjectr   r6   rI   rT   r   r   r   r   r   r   Ú<module>   s0    ![