public function start()
    {
        Loop::run(function () {
            $this->logger->info(sprintf("RPQ is now started, and is listening for new jobs every %d ms", $this->config['poll_interval']), [
                'queue' => $this->queue->getName()
            ]);

            $this->setIsRunning(false);
            Loop::repeat($this->config['poll_interval'], function ($watcherId, $callback) {
                if (!$this->isRunning) {
                    return;
                }

                // Pushes scheduled jobs onto the main queue
                $this->queue->rescheduleJobs();

                // Only allow `max_jobs` to run
                if (count($this->processes) === $this->config['max_jobs']) {
                    return;
                }

                // ZPOP a job from the priority queue
                $job = $this->queue->pop();

                if ($job !== null) {
                    // Spawn a new worker process to handle the job
                    $command = sprintf(
                        'exec %s %s --jobId=%s --name=%s',
                        ($this->config['process']['script'] ?? $_SERVER["SCRIPT_FILENAME"]),
                        $this->config['process']['command'],
                        $job->getId(),
                        $this->queue->getName()
                    );

                    if ($this->config['process']['config'] === true) {
                        $command .= " --config={$this->args['configFile']}";
                    }

                    $process = new Process($command);
                    $process->start();

                    // Grab the PID and push it onto the process stack
                    $pid = yield $process->getPid();
                    $this->logger->info('Started worker', [
                        'pid' => $pid,
                        'command' => $command,
                        'id' => $job->getId(),
                        'queue' => $this->queue->getName()
                    ]);

                    $this->processes[$pid] = [
                        'process' => $process,
                        'id' => $job->getId()
                    ];

                    // Stream any output from the worker in realtime
                    $stream = $process->getStdout();
                    while ($chunk = yield $stream->read()) {
                        $this->logger->info($chunk, [
                            'pid' => $pid,
                            'jobId' => $job->getId(),
                            'queue' => $this->queue->getName()
                        ]);
                    }

                    // When the job is done, it will emit an exit status code
                    $code = yield $process->join();

                    $this->jobHandler->exit(
                        $job->getId(),
                        $pid,
                        $code,
                        false,
                        $this->config['failed_job_backoff_time']
                    );

                    unset($this->processes[$pid]);
                }
            });

            $this->registerSignals();
        });
    }
public function start()
    {
        Loop::run(function () {
            $this->logger->info(sprintf("RPQ is now started, and is listening for new jobs every %d ms", $this->config['poll_interval']), [
                'queue' => $this->queue->getName()
            ]);

            $this->setIsRunning(false);
            Loop::repeat($this->config['poll_interval'], function ($watcherId, $callback) {
                if (!$this->isRunning) {
                    return;
                }

                // Pushes scheduled jobs onto the main queue
                $this->queue->rescheduleJobs();

                // Only allow `max_jobs` to run
                if (count($this->processes) === $this->config['max_jobs']) {
                    return;
                }

                // ZPOP a job from the priority queue
                $job = $this->queue->pop();

                if ($job !== null) {
                    // Spawn a new worker process to handle the job
                    $command = sprintf(
                        'exec %s %s --jobId=%s --name=%s',
                        ($this->config['process']['script'] ?? $_SERVER["SCRIPT_FILENAME"]),
                        $this->config['process']['command'],
                        $job->getId(),
                        $this->queue->getName()
                    );

                    if ($this->config['process']['config'] === true) {
                        $command .= " --config={$this->args['configFile']}";
                    }

                    $process = new Process($command);
                    $process->start();

                    // Grab the PID and push it onto the process stack
                    $pid = yield $process->getPid();
                    $this->logger->info('Started worker', [
                        'pid' => $pid,
                        'command' => $command,
                        'id' => $job->getId(),
                        'queue' => $this->queue->getName()
                    ]);

                    $this->processes[$pid] = [
                        'process' => $process,
                        'id' => $job->getId()
                    ];

                    // Stream any output from the worker in realtime
                    $stream = $process->getStdout();
                    while ($chunk = yield $stream->read()) {
                        $this->logger->info($chunk, [
                            'pid' => $pid,
                            'jobId' => $job->getId(),
                            'queue' => $this->queue->getName()
                        ]);
                    }

                    // When the job is done, it will emit an exit status code
                    $code = yield $process->join();

                    $this->jobHandler->exit(
                        $job->getId(),
                        $pid,
                        $code,
                        false,
                        $this->config['failed_job_backoff_time']
                    );

                    unset($this->processes[$pid]);
                }
            });

            $this->registerSignals();
        });
    }
public function start ( ) { Loop :: run ( function ( ) { $ this -> logger -> info ( sprintf ( "RPQ is now started, and is listening for new jobs every %d ms" , $ this -> config [ 'poll_interval' ] ) , [ 'queue' => $ this -> queue -> getName ( ) ] ) ; $ this -> setIsRunning ( false ) ; Loop :: repeat ( $ this -> config [ 'poll_interval' ] , function ( $ watcherId , $ callback ) { if ( ! $ this -> isRunning ) { return ; } // Pushes scheduled jobs onto the main queue $ this -> queue -> rescheduleJobs ( ) ; // Only allow `max_jobs` to run if ( count ( $ this -> processes ) === $ this -> config [ 'max_jobs' ] ) { return ; } // ZPOP a job from the priority queue $ job = $ this -> queue -> pop ( ) ; if ( $ job !== null ) { // Spawn a new worker process to handle the job $ command = sprintf ( 'exec %s %s --jobId=%s --name=%s' , ( $ this -> config [ 'process' ] [ 'script' ] ?? $ _SERVER [ "SCRIPT_FILENAME" ] ) , $ this -> config [ 'process' ] [ 'command' ] , $ job -> getId ( ) , $ this -> queue -> getName ( ) ) ; if ( $ this -> config [ 'process' ] [ 'config' ] === true ) { $ command .= " --config={$this->args['configFile']}" ; } $ process = new Process ( $ command ) ; $ process -> start ( ) ; // Grab the PID and push it onto the process stack $ pid = yield $ process -> getPid ( ) ; $ this -> logger -> info ( 'Started worker' , [ 'pid' => $ pid , 'command' => $ command , 'id' => $ job -> getId ( ) , 'queue' => $ this -> queue -> getName ( ) ] ) ; $ this -> processes [ $ pid ] = [ 'process' => $ process , 'id' => $ job -> getId ( ) ] ; // Stream any output from the worker in realtime $ stream = $ process -> getStdout ( ) ; while ( $ chunk = yield $ stream -> read ( ) ) { $ this -> logger -> info ( $ chunk , [ 'pid' => $ pid , 'jobId' => $ job -> getId ( ) , 'queue' => $ this -> queue -> getName ( ) ] ) ; } // When the job is done, it will emit an exit status code $ code = yield $ process -> join ( ) ; $ this -> jobHandler -> exit ( $ job -> getId ( ) , $ pid , $ code , false , $ this -> config [ 'failed_job_backoff_time' ] ) ; unset ( $ this -> processes [ $ pid ] ) ; } } ) ; $ this -> registerSignals ( ) ; } ) ; }
Starts an Amp\Loop event loop and begins polling Redis for new jobs
@return void
Starts an Amp \ Loop event loop and begins polling Redis for new jobs