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);
    }
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
Watching events for a job .