o
    ›¨Êh×	  ã                   @   s`   d Z ddlmZ ddlmZ ddlmZ dZeeƒZ	e	j
e	je	jZ
ZZG dd„ dejƒZd	S )
z-Worker <-> Worker Sync at startup (Bootstep).é    )Ú	bootsteps)Ú
get_loggeré   )ÚEvents)ÚMinglec                       sv   e Zd ZdZd ZefZddhZd‡ fdd„	Zdd„ Z	d	d
„ Z
dd„ Zdd„ Zdd„ Zddd„Zdd„ Zdd„ Z‡  ZS )r   zŸBootstep syncing state with neighbor workers.

    At startup, or upon consumer restart, this will:

    - Sync logical clocks.
    - Sync revoked tasks.

    ÚamqpÚredisFc                    s0   | o|   |j¡| _tƒ j|fd|i|¤Ž d S )NÚwithout_mingle)Úcompatible_transportÚappÚenabledÚsuperÚ__init__)ÚselfÚcr	   Úkwargs©Ú	__class__© úO/var/www/html/env/lib/python3.10/site-packages/celery/worker/consumer/mingle.pyr      s   ÿÿ
ÿzMingle.__init__c                 C   s:   |  ¡ �}|jj| jv W  d   ƒ S 1 sw   Y  d S ©N)Úconnection_for_readÚ	transportÚdriver_typeÚcompatible_transports)r   r   Úconnr   r   r   r
       s   
$ÿzMingle.compatible_transportc                 C   s   |   |¡ d S r   )Úsync)r   r   r   r   r   Ústart$   s   zMingle.startc                    sb   t dƒ ˆ ˆ ¡}|r+t dtdd„ | ¡ D ƒƒƒ ‡ ‡fdd„| ¡ D ƒ t dƒ d S t dƒ d S )Nzmingle: searching for neighborszmingle: sync with %s nodesc                 S   s   g | ]\}}|r|‘qS r   r   )Ú.0ÚreplyÚvaluer   r   r   Ú
<listcomp>,   s    zMingle.sync.<locals>.<listcomp>c                    s"   g | ]\}}|rˆ  ˆ ||¡‘qS r   )Úon_node_reply)r   Únodenamer   ©r   r   r   r   r!   -   s    ÿzmingle: sync completezmingle: all alone)ÚinfoÚ
send_helloÚlenÚitems)r   r   Úrepliesr   r$   r   r   '   s   
ÿÿzMingle.syncc                 C   sD   |j jjd|jd�}|jjj}| |j|j	¡pi }| 
|jd ¡ |S )Ng      ð?)ÚtimeoutÚ
connection)r   ÚcontrolÚinspectr+   Ú
controllerÚstateÚrevokedÚhelloÚhostnameÚ_dataÚpop)r   r   r-   Úour_revokedr)   r   r   r   r&   3   s
   
zMingle.send_helloc              
   C   sd   t d|ƒ z| j|fi |¤Ž W d S  ty   ‚  ty1 } ztd||ƒ W Y d }~d S d }~ww )Nz mingle: processing reply from %szmingle: sync with %s failed: %r)ÚdebugÚsync_with_nodeÚMemoryErrorÚ	ExceptionÚ	exception)r   r   r#   r   Úexcr   r   r   r"   :   s   
€ÿzMingle.on_node_replyNc                 K   s   |   ||¡ |  ||¡ d S r   )Úon_clock_eventÚon_revoked_received)r   r   Úclockr0   r   r   r   r   r7   C   s   zMingle.sync_with_nodec                 C   s&   |r|j j |¡ d S |j j ¡  d S r   )r   r>   ÚadjustÚforward)r   r   r>   r   r   r   r<   G   s   &zMingle.on_clock_eventc                 C   s   |r|j jj |¡ d S d S r   )r.   r/   r0   Úupdate)r   r   r0   r   r   r   r=   J   s   ÿzMingle.on_revoked_received)F)NN)Ú__name__Ú
__module__Ú__qualname__Ú__doc__Úlabelr   Úrequiresr   r   r
   r   r   r&   r"   r7   r<   r=   Ú__classcell__r   r   r   r   r      s    	
	r   N)rE   Úceleryr   Úcelery.utils.logr   Úeventsr   Ú__all__rB   Úloggerr6   r%   r:   ÚStartStopStepr   r   r   r   r   Ú<module>   s    