o
    ›¨ÊhE(  ã                   @   s  d Z ddl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 dd
lmZ dZi Zdd„ ZedƒG dd„ dƒƒ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ƒZG dd„ dƒZdS )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   ‡ fdd„}|S )z5Decorator used to register a new result drainer type.c                    s   | t ˆ < | S ©N)Údrainers)Úcls©Úname© úN/var/www/html/env/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   sv   � |p| j j}t ¡ }	 |rt ¡ | |krt ¡ ‚z| j|||d�V  W n
 tjy/   Y nw |r5|ƒ  |jr:d S q©Nr   ©Útimeout)r   Údrain_eventsÚtimeÚ	monotonicÚ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'   >   ó   zDrainer.wait_for)Nr   NNr   )	Ú__name__Ú
__module__Ú__qualname__Ú__doc__r   r   r   r.   r'   r   r   r   r   r       s    
r   c                       sZ   e Zd ZdZdZdZdd„ Zdd„ Z‡ fdd„Zdd	„ Z	d
d„ Z
dd„ Zddd„Z‡  ZS )ÚgreenletDrainerNc                 C   ó   dS )z,create new self._drain_complete_event objectNr   r   r   r   r   Ú_create_drain_complete_eventG   ó   z,greenletDrainer._create_drain_complete_eventc                 C   r7   )z5raise self._drain_complete_event for wakeup .wait_forNr   r   r   r   r   Ú_send_drain_complete_eventK   r9   z*greenletDrainer._send_drain_complete_eventc                    s<   t ƒ j|i |¤Ž t ¡ | _t ¡ | _t ¡ | _|  ¡  d S r   )Úsuperr   Ú	threadingÚEventÚ_startedÚ_stoppedÚ	_shutdownr8   )r   ÚargsÚkwargs©Ú	__class__r   r   r   O   s
   


zgreenletDrainer.__init__c                 C   sd   | 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:   r8   r&   r"   r@   r   r   r   r   ÚrunV   s   

ÿ
ûzgreenletDrainer.runc                 C   s*   | j  ¡ s|  | j¡| _| j  ¡  d S d S r   )r>   rF   ÚspawnrG   Ú_gr,   r   r   r   r   r   a   s   
þzgreenletDrainer.startc                 C   s"   | j  ¡  |  ¡  | j t¡ d S r   )r?   rE   r:   r@   r,   r	   r   r   r   r   r   f   s   
zgreenletDrainer.stopc                 C   s$   |   ¡  |js| jj|d� d S d S r/   )r   r(   Ú_drain_complete_eventr,   r0   r   r   r   r'   k   s   ÿzgreenletDrainer.wait_forr   )r2   r3   r4   rH   rI   rJ   r8   r:   r   rG   r   r   r'   Ú__classcell__r   r   rC   r   r6   B   s    r6   Úeventletc                   @   ó$   e Zd Zdd„ Zdd„ Zdd„ ZdS )ÚeventletDrainerc                 C   s$   ddl m}m} ||ƒ}|dƒ |S )Nr   )r   rH   )rL   r   rH   )r   Úfuncr   rH   Úgr   r   r   rH   t   s   zeventletDrainer.spawnc                 C   ó   ddl m} |ƒ | _d S ©Nr   )r=   )Úeventlet.eventr=   rJ   ©r   r=   r   r   r   r8   z   ó   z,eventletDrainer._create_drain_complete_eventc                 C   s   | j  ¡  d S r   )rJ   Úsendr   r   r   r   r:   ~   r1   z*eventletDrainer._send_drain_complete_eventN©r2   r3   r4   rH   r8   r:   r   r   r   r   rN   q   ó    rN   Úgeventc                   @   rM   )ÚgeventDrainerc                 C   s    dd l }| |¡}| d¡ |S )Nr   )rY   rH   r   )r   rO   rY   rP   r   r   r   rH   …   s   

zgeventDrainer.spawnc                 C   rQ   rR   )Úgevent.eventr=   rJ   rT   r   r   r   r8   ‹   rU   z*geventDrainer._create_drain_complete_eventc                 C   s   | j  ¡  |  ¡  d S r   )rJ   rE   r8   r   r   r   r   r:   �   s   
z(geventDrainer._send_drain_complete_eventNrW   r   r   r   r   rZ   ‚   rX   rZ   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Úappendr`   r_   Ú_wait_for_pendingÚpopleftÚidÚchildren)r   r]   ra   rB   rc   r^   Ú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_resultri   )r   r]   rn   Ú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Útakeri   ©r   r]   r   r   r   rp   À   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_resultsri   r   Úconsume_from)r   Útask_idr]   rn   ÚconcreteÚweak_r   r   r   rq   Ã   s
   
þz%AsyncBackendMixin._add_pending_resultc                    s    ˆ j j ¡  ‡ ‡fdd„|D ƒS )Nc                    s   g | ]
}ˆ j |ˆd d�‘qS )F)rn   rr   )rs   )Ú.0r]   ©r   rn   r   r   Ú
<listcomp>Ë   s    ÿz9AsyncBackendMixin.add_pending_results.<locals>.<listcomp>)r   ro   r   )r   rc   rn   r   r~   r   Úadd_pending_resultsÉ   s   ÿz%AsyncBackendMixin.add_pending_resultsc                 C   s   |   |j¡ |  |¡ |S r   )Ú_remove_pending_resultri   Úon_result_fulfilledrw   r   r   r   Úremove_pending_resultÎ   s   
z'AsyncBackendMixin.remove_pending_resultc                 C   s   | j D ]}| |d ¡ qd S r   )rx   Úpop©r   rz   Úmappingr   r   r   r�   Ó   s   
ÿz(AsyncBackendMixin._remove_pending_resultc                 C   s   | j  |j¡ d S r   )r   Ú
cancel_forri   rw   r   r   r   r‚   ×   ó   z%AsyncBackendMixin.on_result_fulfilledNc                 K   s.   |   ¡  | j|fi |¤ŽD ]}q|j||d�S )N)ÚcallbackÚ	propagate)rb   rg   Úmaybe_throw)r   r]   r‰   rŠ   rB   rl   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   rg   )r   r]   r"   r+   r�   rB   r   r   r   rg   á   s   ÿþýz#AsyncBackendMixin._wait_for_pendingc                 C   r7   ©NTr   r   r   r   r   Úis_asyncê   r9   zAsyncBackendMixin.is_async)T)FT)FrŽ   ©NNN)r2   r3   r4   r5   r_   rm   rs   rp   rq   r€   rƒ   r�   r‚   rŒ   rg   Ú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Úacceptrx   ru   r�   r   r\   r   r   ro   )r   r’   r“   r”   Úpending_resultsÚpending_messagesr   r   r   r   ò   s   zBaseResultConsumer.__init__c                 K   ó   t ƒ ‚r   ©ÚNotImplementedError)r   Úinitial_task_idrB   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   rz   r   r   r   ry     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   )r\   Ú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+   )ro   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   r]   r"   r+   r�   rB   Ú	prev_on_mrl   r   r   r   rg     s    €
þ
üÿûz$BaseResultConsumer._wait_for_pendingc                 K   r   r   r   )r   r]   r"   rB   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   )rx   Ú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Ústatusrz   r   )r�   r   ÚREADY_STATESr©   r¨   ru   Úputrt   r\   r„   rf   r   )r   Úmetar¦   rz   r]   r\   r^   r   r   r   r¤   7  s$   
ý
ÿ
z"BaseResultConsumer.on_state_changer   )NNr�   )r2   r3   r4   r5   r   r   r   r#   ry   r‡   rŸ   rž   r.   rg   r¡   r§   r©   r¤   r   r   r   r   r   ï   s"    


ÿ
r   )r5   r&   r<   r$   Úcollectionsr   Úqueuer   r   Úweakrefr   Úkombu.utils.compatr   Úceleryr   Úcelery.exceptionsr   Úcelery.utils.threadsr	   Ú__all__r   r   r   r6   rN   rZ   r
   r   r   r   r   r   Ú<module>   s0    !/[