src/Subscribers/WatchdogSubscriber.php
public function watchdog(string $jid, string $worker, ?int $pid = null)
    {
        if (empty($this->channels) || $pid === null) {
            return;
        }

        ini_set('default_socket_timeout', self::UNLIMITED);

        /**
         * Initialize a new pubsub consumer.
         *
         * @var Consumer $pubsub|null
         */
        $pubsub = $this->redis->pubSubLoop();

        $callable = [$pubsub, 'subscribe'];
        if (!is_callable($callable)) {
            return;
        }

        call_user_func_array($callable, $this->channels);

        /** @var \stdClass $message */
        foreach ($pubsub as $message) {
            if ($message->kind !== 'message' || empty($message->payload)) {
                continue;
            }

            $payload = json_decode($message->payload, true);
            if (empty($payload)) {
                continue;
            }

            if (empty($payload['event']) || !is_array($payload)) {
                continue;
            }

            if (!in_array($payload['event'], self::WATCHDOG_EVENTS, true) || empty($payload['jid'])) {
                continue;
            }

            if ($payload['jid'] !== $jid) {
                continue;
            }

            $who = 'watchdog:' . $worker;

            switch ($payload['event']) {
                case self::LOCK_LOST:
                    if (!empty($payload['worker']) && $payload['worker'] === $worker) {
                        $this->logger->info(
                            "{type}: sending SIGKILL to child {$pid}; job {jid} handed out to another worker",
                            ['type' => $who, 'jid' => $jid]
                        );

                        $this->system->posixKill($pid, SIGKILL);
                        $pubsub->stop();
                    }
                    break;
                case self::CANCELED:
                    if (!empty($payload['worker']) && $payload['worker'] === $worker) {
                        $this->logger->info(
                            "{type}: sending SIGKILL to child {$pid}; job {jid} canceled",
                            ['type' => $who, 'jid' => $jid]
                        );
                        $this->system->posixKill($pid, SIGKILL);
                        $pubsub->stop();
                    }
                    break;
                case self::COMPLETED:
                case self::FAILED:
                    $pubsub->stop();
                    break;
            }
        }

        // Always unset the pubsub consumer instance when you are done! The
        // class destructor will take care of cleanups and prevent protocol
        // desynchronizations between the client and the server.
        unset($pubsub);

        ini_set('default_socket_timeout', $this->defaultSocketTimeout);
    }
public function watchdog(string $jid, string $worker, ?int $pid = null)
    {
        if (empty($this->channels) || $pid === null) {
            return;
        }

        ini_set('default_socket_timeout', self::UNLIMITED);

        /**
         * Initialize a new pubsub consumer.
         *
         * @var Consumer $pubsub|null
         */
        $pubsub = $this->redis->pubSubLoop();

        $callable = [$pubsub, 'subscribe'];
        if (!is_callable($callable)) {
            return;
        }

        call_user_func_array($callable, $this->channels);

        /** @var \stdClass $message */
        foreach ($pubsub as $message) {
            if ($message->kind !== 'message' || empty($message->payload)) {
                continue;
            }

            $payload = json_decode($message->payload, true);
            if (empty($payload)) {
                continue;
            }

            if (empty($payload['event']) || !is_array($payload)) {
                continue;
            }

            if (!in_array($payload['event'], self::WATCHDOG_EVENTS, true) || empty($payload['jid'])) {
                continue;
            }

            if ($payload['jid'] !== $jid) {
                continue;
            }

            $who = 'watchdog:' . $worker;

            switch ($payload['event']) {
                case self::LOCK_LOST:
                    if (!empty($payload['worker']) && $payload['worker'] === $worker) {
                        $this->logger->info(
                            "{type}: sending SIGKILL to child {$pid}; job {jid} handed out to another worker",
                            ['type' => $who, 'jid' => $jid]
                        );

                        $this->system->posixKill($pid, SIGKILL);
                        $pubsub->stop();
                    }
                    break;
                case self::CANCELED:
                    if (!empty($payload['worker']) && $payload['worker'] === $worker) {
                        $this->logger->info(
                            "{type}: sending SIGKILL to child {$pid}; job {jid} canceled",
                            ['type' => $who, 'jid' => $jid]
                        );
                        $this->system->posixKill($pid, SIGKILL);
                        $pubsub->stop();
                    }
                    break;
                case self::COMPLETED:
                case self::FAILED:
                    $pubsub->stop();
                    break;
            }
        }

        // Always unset the pubsub consumer instance when you are done! The
        // class destructor will take care of cleanups and prevent protocol
        // desynchronizations between the client and the server.
        unset($pubsub);

        ini_set('default_socket_timeout', $this->defaultSocketTimeout);
    }
Watching events for a job.

@param  string   $jid
@param  string   $worker
@param  int|null $pid
@return void