File: //usr/local/lib/python3.9/site-packages/celery/contrib/testing/__pycache__/worker.cpython-39.pyc
a
X>h1 � @ s` d Z ddlZddlZddlZddlmZ ddlmZmZm Z m
Z
ddl Z ddl m
Z
mZ ddlmZmZ ddlmZ ddlmZ ej�d d
�Zed i d �Zed
ddhd �Zeddhd �ZG dd� dej �Z eddeddddfdd��Z eddede ddfe
e e e
e e f e e ee!e"eej d� dd��Z#eddedfd d ��Z$e
e
e e f e dd �d d �Z%dS )!z'Embedded workers for integration tests.� N)�contextmanager)�Any�Iterable�Optional�Union)�Celery�worker)�_set_task_join_will_block�allow_join_result)�Signal)�
anon_nodename�WORKER_LOGLEVEL�error�test_worker_starting)�nameZproviding_args�test_worker_startedr �consumer�test_worker_stoppedc sX e Zd ZdZdZdZ� fdd�ZG dd� dejj �Z � fdd �Z
d
d � Z d d
� Z � Z
S )�TestWorkControllerz3Worker that can synchronize on being fully started.FNc s� t �� | _t� j|i |�� | jj�d�d dkr�ddlm } |� | _
t � � | _
zddlm} |�� W n tyz Y n0 tj�| j
t�� �| _| j�� d S )N�.���Zpreforkr )�Queue)�pickling_support)� threading�Event� _on_started�super�__init__�pool_cls�
__module__�splitZbilliardr � logger_queue�os�getpid�pidZtblibr �install� ImportError�logging�handlers�
QueueListener� getLoggerZqueue_listener�start)�self�args�kwargsr r �� __class__� �G/usr/local/lib/python3.9/site-packages/celery/contrib/testing/worker.pyr ' s
zTestWorkController.__init__c @ s e Zd Zdd� Zdd� ZdS )z TestWorkController.QueueHandlerc C s
d|_ |S )NT)�
from_queue�r, �recordr1 r1 r2 �prepare> s z'TestWorkController.QueueHandler.preparec C s t jr� d S )N)r' �raiseExceptionsr4 r1 r1 r2 � handleErrorC s z+TestWorkController.QueueHandler.handleErrorN)�__name__r � __qualname__r6 r8 r1 r1 r1 r2 � QueueHandler= s r; c s@ � j r6� �� j �}|�� fdd�� t�� }|�|� t� �� S )Nc s | j � jkot| dd� S )Nr3 F)�processr$ �getattr)�r�r, r1 r2 �<lambda>J � z*TestWorkController.start.<locals>.<lambda>)r! r; � addFilterr' r* �
addHandlerr r+ )r, �handler�loggerr/ r? r2 r+ G s
zTestWorkController.startc C s | j �� tj| j| |d� dS )z=Callback called when the Consumer blueprint is fully started.)�senderr r N)r �setr �send�app)r, r r1 r1 r2 �on_consumer_readyO s
�z$TestWorkController.on_consumer_readyc C s | j �� dS )z�Wait for worker to be fully up and running.
Warning:
Worker must be started within a thread for this to work,
or it will block forever.
N)r �waitr? r1 r1 r2 �ensure_startedV s z!TestWorkController.ensure_started)r9 r r: �__doc__Z__test__r! r r' r( r; r+ rJ rL �
__classcell__r1 r1 r/ r2 r s
r � ZsoloTg $@c
k s� t j| d� d} z�t| f||||||d�|���d} |r�ddlm}
t� �( |
�� j|d�dksbJ �W d � n1 sv0 Y | V W d � n1 s�0 Y W tj| | d� ntj| | d� 0 dS ) z[Start embedded worker.
Yields:
celery.app.worker.Worker: worker instance.
)rF N)� concurrency�pool�loglevel�logfile�perform_ping_check�shutdown_timeoutrO )�ping)�timeoutZpong)rF r ) r rH �_start_worker_thread�tasksrV r
�delay�getr ) rI rP rQ rR rS rT Zping_task_timeoutrU r. r rV r1 r1 r2 � start_workera s( �� 6&r\ ) rI rP rQ rR rS �WorkControllerrT rU �returnc
k s0 t | ||� |r d| jv s J �| jtj�d�d��} | jj W d � n1 sP0 Y |f | ||�dt � �|||d|�dd�ddd�
|��}
t
j |
j dd �} | � � |
�
� td
� z<|
V W d d lm} d | _| �|� | �� r�td
��d| _n6d d lm} d | _| �|� | �� �r$td
��d| _0 dS )zaStart Celery worker in a thread.
Yields:
celery.worker.Worker: worker instance.
z celery.pingZ TEST_BROKER)�hostnameNr_ �without_heartbeatT)
rI rP r_ rQ rR rS Zready_callbackr` Zwithout_mingleZwithout_gossip)�target�daemonFr )�statez�Worker thread failed to exit within the allocated timeout. Consider raising `shutdown_timeout` if your tasks take longer to execute.)�setup_app_for_workerrY �
connectionr" �environr[ Zdefault_channelZ
queue_declare�popr r �Threadr+ rL r Z
celery.workerrc Zshould_terminate�join�is_alive� RuntimeError)
rI rP rQ rR rS r] rT rU r. �connr �trc r1 r1 r2 rX � sR &
� �
��
�rX c k sP ddl m}m} | �� ||d�g�}|�� zdV W |�� n
|�� 0 dS )zfStart worker in separate process.
Yields:
celery.app.worker.Worker: worker instance.
r )�Cluster�Nodeztestworker1@%hN)Zcelery.apps.multirn ro � set_currentr+ Zstopwait) rI rP rQ rR rS r. rn ro Zclusterr1 r1 r2 �_start_worker_process� s
rq )rI rR rS r^ c C s8 | � � | �� | �� dt| j�_| jj||d� dS )z9Setup the app to be used for starting an embedded worker.F)rR rS N)�finalizerp � set_default�type�log�_setup�setup)rI rR rS r1 r1 r2 rd � s
rd )&rM r' r" r �
contextlibr �typingr r r r Zcelery.worker.consumerZceleryr r Z
celery.resultr r
Zcelery.utils.dispatchr Zcelery.utils.nodenamesr rf r[ r
r r r r] r r\ �int�str�bool�floatrX rq rd r1 r1 r1 r2 �<module> st ���C� '�
�7�