o /i~@s,ddlZddlZddlZddlmZmZer"ddlmZddlm Z ddl m Z ddl m Z dZd d d ed efd dZdeeefdeeeffddZd d d efddZd d d efddZd$d d defddZdddeeeffddZd%ddZdddeeeffd d!Zdddeeeffd"d#ZdS)&N) TYPE_CHECKINGAny)Redis) BaseWorker)InvalidJobOperation)Jobz rq:pubsub:%s connectionr worker_namecommandcKs0d|i}|r |||t|t|dS)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 kwargspayloadrJ/var/www/html/flask_server/venv/lib/python3.10/site-packages/rq/command.py send_commands rrreturncCst|dS)zd Returns a dict of command data Args: payload (dict): Parses the payload dict. data)rloadsdecode)rrrr parse_payload&srcCt||ddS)z Sends a command to shutdown a worker. Args: connection (Redis): A Redis Connection worker_name (str): The Job ID shutdownNrr r rrrsend_shutdown_command0rcCr)z Tell worker to kill it's horse Args: connection (Redis): A Redis Connection worker_name (str): The Job ID kill-horseNrrrrrsend_kill_horse_command;r r"job_idcCs4tj|||d}|jstdt||jd|ddS)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)rfetchr rr)r r#r$jobrrrsend_stop_job_commandFs r(workerrcCsP|ddkr t||dS|ddkrt|dS|ddkr&t||dSdS)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)rrrrhandle_commandUs    r.cCs&|jdt}t|tjdS)zUPerform shutdown command. Args: worker (Worker): The worker to use. z1Received shutdown command, sending SIGINT signal.N)loginfoosgetpidkillsignalSIGINT)r)pidrrrr+ds r+cCs:|jd|jr|jd|dS|jddS)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)r/r0 horse_pid kill_horser-rrrr,os   r,cCsL|d}|jd||r||kr||_|dS|jd|dS)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)getr/debugget_current_job_id_stopped_job_idr8r0)r)rr#rrrr*s  r*)N)r)r)rr1r4typingrrredisrr)r rq.exceptionsrrq.jobrrstrrdictrrr"r(r.r+r,r*rrrrs&    "