
    xaiF,                     n   d Z ddl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dZ G d d      Zi Zd Z ed       G d d             Z G d de      Z ed       G d de             Z ed       G d de             Z G d d      Z G d d      Zy)z$Async I/O backend support utilities.    N)deque)Empty)sleep)WeakKeyDictionary)detect_environment)states)TimeoutError)THREAD_TIMEOUT_MAXz@Celery must be restarted because a shutdown signal was detected.)AsyncBackendMixinBaseResultConsumerDrainerregister_drainerc                   *    e Zd ZdZd Zd Zd ZddZy)EventletAdaptedEventzq
    An adapted eventlet event, designed to match the API of `threading.Event` and
    `gevent.event.Event`.
    c                 6    dd l }|j                         | _        y Nr   )eventletEventevt)selfr   s     n/var/www/html/BankruptcyAI-uat/bankruptcy-ai/venv/lib/python3.12/site-packages/celery/backends/asynchronous.py__init__zEventletAdaptedEvent.__init__    s    >>#    c                 6    | j                   j                         S N)r   readyr   s    r   is_setzEventletAdaptedEvent.is_set$   s    xx~~r   c                 6    | j                   j                         S r   )r   sendr   s    r   setzEventletAdaptedEvent.set'   s    xx}}r   Nc                 8    | j                   j                  |      S r   )r   waitr   timeouts     r   r#   zEventletAdaptedEvent.wait*   s    xx}}W%%r   r   )__name__
__module____qualname____doc__r   r   r!   r#    r   r   r   r      s    
$ &r   r   c                       fd}|S )z5Decorator used to register a new result drainer type.c                     | t         <   | S r   )drainers)clsnames    r   _innerz register_drainer.<locals>._inner3   s    
r   r*   )r/   r0   s   ` r   r   r   1   s     Mr   defaultc                   8    e Zd ZdZd Zd Zd Zd	dZd
dZd Z	y)r   zResult draining service.c                     || _         y r   )result_consumer)r   r4   s     r   r   zDrainer.__init__=   s
    .r   c                      y r   r*   r   s    r   startzDrainer.start@       r   c                      y r   r*   r   s    r   stopzDrainer.stopC   r7   r   Nc              #   V  K   |xs | j                   j                  }t        j                         }	 |r.t        j                         |z
  |k\  rt	        j
                         	 | j                  |||       |r |        |j                  ry ^# t        j
                  $ r Y ,w xY wwNr%   )r4   drain_eventstime	monotonicsocketr%   wait_forr   )r   pr%   intervalon_intervalr#   
time_starts          r   drain_events_untilzDrainer.drain_events_untilF   s     8t++88^^%
4>>+j8GCnn&&mmAtXm>> ww  >> s*   A B)#B 9B)B&#B)%B&&B)c                      ||       y r;   r*   r   rB   r#   r%   s       r   rA   zDrainer.wait_forW   s
    Wr   c                 *    t        j                         S r   )	threadingr   r   s    r   _eventzDrainer._eventZ   s      r   )N   NNr   )
r&   r'   r(   r)   r   r6   r9   rF   rA   rK   r*   r   r   r   r   9   s#    "/"!r   r   c                   T     e Zd ZdZdZdZdZd Z fdZd Z	d Z
d Zd	dZd Z xZS )
greenletDrainerNc                 b    | j                   j                          | j                         | _         y r   )_drain_complete_eventr!   rK   r   s    r   _send_drain_complete_eventz*greenletDrainer._send_drain_complete_eventd   s"    ""&&(%)[[]"r   c                     t        |   |i | | j                         | _        | j                         | _        | j                         | _        | j                         | _        y r   )superr   rK   _started_stopped	_shutdownrP   )r   argskwargs	__class__s      r   r   zgreenletDrainer.__init__h   sJ    $)&)%)[[]"r   c                    | j                   j                          	 | j                  j                         sH	 | j                  j                  d       | j                          | j                  j                         sH| j                          	 | j                  j                          y # t        j                  $ r Y \w xY w# t        $ r}|| _
         d }~ww xY w# t        $ r"}t        j                  d|        Y d }~y d }~ww xY w# | j                          	 | j                  j                          w # t        $ r"}t        j                  d|        Y d }~w d }~ww xY wxY w)NrL   r<   zFailed to set shutdown event: )rT   r!   rU   r   r4   r=   rQ   r@   r%   	Exception_excrV   RuntimeErrorloggingerror)r   es     r   runzgreenletDrainer.runp   s.   	Dmm**,((55a5@335 mm**, ++-D""$ ~~  	DI	   D >qcBCCD ++-D""$ D >qcBCCDs   C ,B* #C C *C =C ?C  C 	CCCD
 	D%DD
E%D76E%7	E" EE%E""E%c                     | j                          | j                  j                         s;| j                  | j                        | _        | j                  j                          y y r   )_ensure_not_shut_downrT   r   spawnra   _gr#   r   s    r   r6   zgreenletDrainer.start   sG    ""$}}##%jj*DGMM  &r   c                 v    | j                   j                          | j                  j                  t               y r   )rU   r!   rV   r#   r
   r   s    r   r9   zgreenletDrainer.stop   s%    ./r   c                     | j                          |j                  s-| j                  j                  |       | j	                          y y r;   )r6   r   rP   r#   rc   rH   s       r   rA   zgreenletDrainer.wait_for   s:    

ww&&++G+<&&( r   c                     | j                   j                         r'| j                  | j                  t        t              y)aJ  Currently used to ensure the drainer has not run to completion.

        Raises if the shutdown event has been signaled (either due to an exception
        or stop() being called).

        The _shutdown event acts as synchronization to ensure _exc is properly
        set before it is read from, avoiding need for locks.
        N)rV   r   r\   r[   E_CELERY_RESTART_REQUIREDr   s    r   rc   z%greenletDrainer._ensure_not_shut_down   s8     >>  "yy$ii 9::	 #r   r   )r&   r'   r(   rd   r\   re   rP   rQ   r   ra   r6   r9   rA   rc   __classcell__)rY   s   @r   rN   rN   ^   s<    ED	B 33D(!0);r   rN   r   c                       e Zd Zd Zd Zy)eventletDrainerc                 6    ddl m}m}  ||      } |d       |S )Nr   )r   rd   )r   r   rd   )r   funcr   rd   gs        r   rd   zeventletDrainer.spawn   s    )$Kar   c                     t               S r   )r   r   s    r   rK   zeventletDrainer._event   s    #%%r   Nr&   r'   r(   rd   rK   r*   r   r   rl   rl      s    &r   rl   geventc                       e Zd Zd Zd Zy)geventDrainerc                 R    dd l }|j                  |      }|j                  d       |S r   )rr   rd   r   )r   rn   rr   ro   s       r   rd   zgeventDrainer.spawn   s#    LLQr   c                     ddl m}  |       S )Nr   )r   )gevent.eventr   )r   r   s     r   rK   zgeventDrainer._event   s    &wr   Nrq   r*   r   r   rt   rt      s    r   rt   c                   r    e Zd ZdZd ZddZddZd ZddZddZ	d Z
d	 Zd
 Z	 ddZ	 ddZed        Zy)r   z.Mixin for backends that enables the async API.c                 6    || j                   j                  |<   y r   )r4   buckets)r   resultbuckets      r   _collect_intozAsyncBackendMixin._collect_into   s    /5$$V,r   c              +   \  K   | j                          |j                  }|s
t               t               }|D ]P  }t	        |d      s|j                  |       !|j                  r|j                  |       ?| j                  ||       R  | j                  |fd|i|D ]Y  }|s|j                         }t	        |d      s|j                  |j                  f n|j                  |j                  f |rT[ |r.|j                         }|j                  |j                  f |r-y y w)N_cacheno_ack)_ensure_not_eagerresultsStopIterationr   hasattrappendr   r}   _wait_for_pendingpopleftidchildren)r   r{   r   rX   r   r|   node_s           r   iter_nativezAsyncBackendMixin.iter_native   s     ../!  	1D4*d#d#""40	1 (''HvHH 	/A~~'tX.''4==00''4;;.. 	/ >>#D''4;;&& s   B"D,%AD,90D,*D,c                     |r$| j                   j                  j                          	 | j                  |       |S # t        $ r" | j                  |j                  ||       Y |S w xY w)N)weak)r4   drainerr6   _maybe_resolve_from_bufferr   _add_pending_resultr   )r   r{   r   start_drainers       r   add_pending_resultz$AsyncBackendMixin.add_pending_result   si      ((..0	C++F3   	C$$VYYT$B	Cs   ; 'A&%A&c                 l    |j                  | j                  j                  |j                               y r   )_maybe_set_cache_pending_messagestaker   r   r{   s     r   r   z,AsyncBackendMixin._maybe_resolve_from_buffer   s%     6 6 ; ;FII FGr   c                     | j                   \  }}||vr4|j                  |vr%||r|n||<   | j                  j                  |       y y y r   )_pending_resultsr   r4   consume_from)r   task_idr{   r   concreteweak_s         r   r   z%AsyncBackendMixin._add_pending_result   sN    //%%FIIX$=5;dU'2  --g6 %>r   c                     | j                   j                  j                          |D cg c]  }| j                  ||d       c}S c c}w )NF)r   r   )r4   r   r6   r   )r   r   r   r{   s       r   add_pending_resultsz%AsyncBackendMixin.add_pending_results   sL    $$**,%' ''T'O ' 	' 's   Ac                 ^    | j                  |j                         | j                  |       |S r   )_remove_pending_resultr   on_result_fulfilledr   s     r   remove_pending_resultz'AsyncBackendMixin.remove_pending_result   s'    ##FII.  (r   c                 J    | j                   D ]  }|j                  |d         y r   )r   popr   r   mappings      r   r   z(AsyncBackendMixin._remove_pending_result   s%    ,, 	'GKK&	'r   c                 N    | j                   j                  |j                         y r   )r4   
cancel_forr   r   s     r   r   z%AsyncBackendMixin.on_result_fulfilled  s    ''		2r   Nc                 x    | j                           | j                  |fi |D ]  } |j                  ||      S )N)callback	propagate)r   r   maybe_throw)r   r{   r   r   rX   r   s         r   wait_for_pendingz"AsyncBackendMixin.wait_for_pending  sH     '''9&9 	A	!!8y!IIr   c                 D     | j                   j                  |f|||d|S )N)r%   rD   
on_message)r4   r   )r   r{   r%   rD   r   rX   s         r   r   z#AsyncBackendMixin._wait_for_pending  s9     6t##55
##

 
 	
r   c                      yNTr*   r   s    r   is_asynczAsyncBackendMixin.is_async  s    r   )T)FT)Fr   NNN)r&   r'   r(   r)   r}   r   r   r   r   r   r   r   r   r   r   propertyr   r*   r   r   r   r      s^    86':H7'

'3 37J FJ
  r   r   c                   n    e Zd ZdZd Zd Zd ZddZd Zd Z	d	 Z
d
 ZddZ	 ddZddZd Zd Zd Zy)r   z2Manager responsible for consuming result messages.c                     || _         || _        || _        || _        || _        d | _        t               | _        t        t                  |       | _
        y r   )backendappacceptr   r   r   r   rz   r-   r   r   )r   r   r   r   pending_resultspending_messagess         r   r   zBaseResultConsumer.__init__  sM     /!1(* 2 45d;r   c                     t               r   NotImplementedError)r   initial_task_idrX   s      r   r6   zBaseResultConsumer.start*      !##r   c                      y r   r*   r   s    r   r9   zBaseResultConsumer.stop-  r7   r   Nc                     t               r   r   r$   s     r   r=   zBaseResultConsumer.drain_events0  r   r   c                     t               r   r   r   r   s     r   r   zBaseResultConsumer.consume_from3  r   r   c                     t               r   r   r   s     r   r   zBaseResultConsumer.cancel_for6  r   r   c                     | j                   j                          t               | _         d | _        | j	                          y r   )rz   clearr   r   on_after_forkr   s    r   _after_forkzBaseResultConsumer._after_fork9  s/    (*r   c                      y r   r*   r   s    r   r   z BaseResultConsumer.on_after_fork?  r7   r   c                 >    | j                   j                  |||      S )Nr%   rD   )r   rF   )r   rB   r%   rD   s       r   rF   z%BaseResultConsumer.drain_events_untilB  s%    ||..wK / 9 	9r   c              +   0  K    | j                   |fd|i| | j                  |c}| _        	 | j                  |j                  ||      D ]  }d  t	        d        	 || _        y # t
        j                  $ r t        d      w xY w# || _        w xY ww)Nr%   r   r   zThe operation timed out.)on_wait_for_pendingr   rF   on_readyr   r@   r%   r	   )r   r{   r%   rD   r   rX   	prev_on_mr   s           r   r   z$BaseResultConsumer._wait_for_pendingF  s      	!  CCFC%)__j"	4?		(,,OOW + - -  a	 (DO ~~ 	;9::	; (DOs(   *B2A(  B(BB
 
	BBc                      y r   r*   )r   r{   r%   rX   s       r   r   z&BaseResultConsumer.on_wait_for_pendingV  r7   r   c                 <    | j                  |j                  |       y r   )on_state_changepayload)r   messages     r   on_out_of_band_resultz(BaseResultConsumer.on_out_of_band_resultY  s    W__g6r   c                 f    | j                   D ]	  }	 ||   c S  t        |      # t        $ r Y "w xY wr   )r   KeyErrorr   s      r   _get_pending_resultz&BaseResultConsumer._get_pending_result\  sH    ,, 	Gw''	
 w  s   $	00c                    | j                   r| j                  |       |d   t        j                  v rW|d   }	 | j                  |      }|j	                  |       | j
                  }	 |j                  |      }|j                  |       t        d       y # t        $ r Y w xY w# t        $ r | j                  j                  ||       Y Bw xY w)Nstatusr   r   )r   r   READY_STATESr   r   rz   r   r   r   r   putr   )r   metar   r   r{   rz   r|   s          r   r   z"BaseResultConsumer.on_state_changed  s    ??OOD!>V0009oG*11': ''-,,*$[[0F
 MM&)a     : &&**7D9:s#   B% (B 	B"!B"%%CCr   )NNr   )r&   r'   r(   r)   r   r6   r9   r=   r   r   r   r   rF   r   r   r   r   r   r*   r   r   r   r     sR    <	<$$$$9
 FJ( 7 r   r   )r)   r^   r@   rJ   r>   collectionsr   queuer   r   weakrefr   kombu.utils.compatr   celeryr   celery.exceptionsr	   celery.utils.threadsr
   ri   __all__r   r-   r   r   rN   rl   rt   r   r   r*   r   r   <module>r      s    *        % 1  * 3^ & &(  )!! !! !!HE;g E;P *	&o 	& 	& (
O 
 
X Xv^ ^r   