o
    h                     @   s   d 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ZeeZdd
dZdddZ	dddZ	dddZdS )z*The consumers highly-optimized inner loop.    N)	bootsteps)WorkerLostError)
get_logger   )state)asynloopsynloop皙?c              
   C   sh   z	| j |d W d S  ty3 } zt|dd }|d ur!|tjkr( W Y d }~d S W Y d }~d S d }~ww )Ntimeouterrno)drain_events	Exceptiongetattrr   EAGAIN)
connectionr   exc	exc_errno r   U/var/www/html/optinet_system/venv/lib/python3.10/site-packages/celery/worker/loops.py_quick_drain   s   r          @c                    sJ   d g sS    }|r jsS  fdd}| || ||f S )Nc              
      s@   z  |  W d S  ty } z|d< W Y d }~d S d }~ww )Nr   )heartbeat_checkr   )rateer   heartbeat_errorr   r   tick&   s   z#_enable_amqheartbeats.<locals>.tick)get_heartbeat_intervalsupports_heartbeatscall_repeatedly)timerr   r   	heartbeatr   r   r   r   _enable_amqheartbeats   s   
r#   c	                 C   s  t j}	|j}
|j}|  }t|j||d}||_| j	| | 	| |
  |   | js7| j s7td|jjdkrC|t| ||_| }z]|j|	kr| jrt  |d dura|d |j|jkrj|
  zt| W n ty}   | }Y nw |j|	kr| jsSW z|  W dS  ty } zt d| W Y d}~dS d}~ww z|  W w  ty } zt d| W Y d}~w d}~ww )zNon-blocking event loop.r   z Could not start worker processesamqpr   Nz&Error cleaning up after event loop: %r)!r   RUNupdateconnection_errorscreate_task_handlerr#   r!   
on_message
controllerregister_with_event_loopconsumeon_readyrestart_countpooldid_start_okr   	transportdriver_type	call_soonr   propagate_errorscreate_loopr   r   maybe_shutdownprevvaluenextStopIterationresetr   logger	exception)objr   consumer	blueprinthubqosr"   clockhbrater&   
update_qoserrorson_task_receivedr   loopr   r   r   r   r   2   s\   
r   c	                    s   t j |  }
| jdgt| jddrt| j|d|
|_|	  | 
   fdd}j krR| jrTz	t  W |  n|  w j krV| js:dS dS dS dS )zEFallback blocking event loop for transports that doesn't support AIO.Nis_greenFr$   c                      sv   d dur
d j jkr  z  jdd W dS  tjy+   Y dS  ty:   j kr7 Y dS w )zC
        Perform one iteration of the blocking event loop.
        r   Nr   r
   )r8   r9   r'   r   socketr   OSErrorr   r   r&   rA   r   r   perform_pending_operationsrC   r   r   _loop_cyclez   s   
zsynloop.<locals>._loop_cycle)r   r&   r)   rN   r   r0   r#   r!   r*   r-   r.   r   r   r7   )r?   r   r@   rA   rB   rC   r"   rD   rE   kwargsrH   rO   r   rM   r   r   l   s   
 r   )r	   )r   )__doc__r   rK   celeryr   celery.exceptionsr   celery.utils.logr    r   __all____name__r=   r   r#   r   r   r   r   r   r   <module>   s    

	
;