]j~$ddlZddlZddlZddlmZmZer ddlmZddlm Z ddl m Z ddl m Z dZd d d ed efd ZdeeefdeeeffdZd d d efdZd d d efdZdd d defdZdddeeeffdZddZdddeeeffdZdddeeeffdZy)N) TYPE_CHECKINGAny)Redis) BaseWorker)InvalidJobOperation)Jobz rq:pubsub:%s connectionr worker_namecommandc d|i}|r|j||jt|ztj|y)a$ Sends a command to a worker. A command is just a string, available commands are: - `shutdown`: Shuts down a worker - `kill-horse`: Command for the worker to kill the current working horse - `stop-job`: A command for the worker to stop the currently running job The command string will be parsed into a dictionary and send to a PubSub Topic. Workers listen to the PubSub, and `handle` the specific command. Args: connection (Redis): A Redis Connection worker_name (str): The Job ID r N)updatepublishPUBSUB_CHANNEL_TEMPLATEjsondumps)r r r kwargspayloads 4/root/env/lib/python3.12/site-packages/rq/command.py send_commandrs<'"G v.QRrreturncNtj|djS)zd Returns a dict of command data Args: payload (dict): Parses the payload dict. data)rloadsdecode)rs r parse_payloadr&s  ::gfo,,. //rct||dy)z Sends a command to shutdown a worker. Args: connection (Redis): A Redis Connection worker_name (str): The Job ID shutdownNrr r s rsend_shutdown_commandr"0s[*5rct||dy)z Tell worker to kill it's horse Args: connection (Redis): A Redis Connection worker_name (str): The Job ID kill-horseNr r!s rsend_kill_horse_commandr%;s[,7rjob_idctj|||}|js tdt ||jd|y)z Instruct a worker to stop a job Args: connection (Redis): A Redis Connection job_id (str): The Job ID serializer (): The serializer )r serializerzJob is not currently executingstop-job)r&N)r fetchr rr)r r&r(jobs rsend_stop_job_commandr,Fs; ))Fzj IC ??!"BCCS__jHrworkerrc|ddk(r t||y|ddk(r t|y|ddk(r t||yy)zParses payload and routes commands to the worker. Args: worker (Worker): The worker to use payload (Dict[Any, Any]): The Payload r r)rr$N)handle_stop_job_commandhandle_shutdown_commandhandle_kill_worker_commandr-rs rhandle_commandr3UsLyZ'0  z )'  | +"673 ,rc|jjdtj}tj|t j y)zUPerform shutdown command. Args: worker (Worker): The worker to use. z1Received shutdown command, sending SIGINT signal.N)loginfoosgetpidkillsignalSIGINT)r-pids rr0r0ds2  JJOOGH ))+CGGCrc|jjd|jr,|jjd|jy|jjdy)z Stops work horse Args: worker (Worker): The worker to stop payload (Dict[Any, Any]): The payload. zReceived kill horse command.zKilling horse...z1Worker is not working, kill horse command ignoredN)r5r6 horse_pid kill_horser2s rr1r1osL JJOO23  *+ KLrc|jd}|jjd||r+|j|k(r||_|j y|jj d|y)zHandles stop job command. Args: worker (Worker): The worker to use payload (Dict[Any, Any]): The payload. r&zReceived command to stop job %sz'Not working on job %s, command ignored.N)getr5debugget_current_job_id_stopped_job_idr?r6)r-rr&s rr/r/sc[[ "F JJ6? &++-7"( A6Jr)N)r-r)rr7r:typingrrredisrr-r rq.exceptionsrrq.jobr rstrrdictrr"r%r,r3r0r1r/rrrLs %"-(SWS3SS*04S>0d38n06g6C688c8 Ig Is I 4< 4$sCx. 4 M|Md38nM"KLK4S>Kr