src/Process/Dispatcher.php
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