protected function processLocalQueue(): bool
    {
        $now = time();
        foreach ($this->queue as $key => $job) {
            $this->db->{$this->scheduler->getJobQueue()}->updateOne(
                ['_id' => $job['_id'], '$isolated' => true],
                ['$setOnInsert' => $job],
                ['upsert' => true]
            );

            if ($job['options']['at'] <= $now) {
                $this->logger->info('postponed job ['.$job['_id'].'] ['.$job['class'].'] can now be executed', [
                    'category' => get_class($this),
                    'pm' => $this->process,
                ]);

                unset($this->queue[$key]);
                $job['options']['at'] = 0;

                if (true === $this->collectJob($job, JobInterface::STATUS_PROCESSING, JobInterface::STATUS_POSTPONED)) {
                    $this->processJob($job);
                }
            }
        }

        return true;
    }
protected function processLocalQueue(): bool
    {
        $now = time();
        foreach ($this->queue as $key => $job) {
            $this->db->{$this->scheduler->getJobQueue()}->updateOne(
                ['_id' => $job['_id'], '$isolated' => true],
                ['$setOnInsert' => $job],
                ['upsert' => true]
            );

            if ($job['options']['at'] <= $now) {
                $this->logger->info('postponed job ['.$job['_id'].'] ['.$job['class'].'] can now be executed', [
                    'category' => get_class($this),
                    'pm' => $this->process,
                ]);

                unset($this->queue[$key]);
                $job['options']['at'] = 0;

                if (true === $this->collectJob($job, JobInterface::STATUS_PROCESSING, JobInterface::STATUS_POSTPONED)) {
                    $this->processJob($job);
                }
            }
        }

        return true;
    }
protected function processLocalQueue ( ) : bool { $ now = time ( ) ; foreach ( $ this -> queue as $ key => $ job ) { $ this -> db -> { $ this -> scheduler -> getJobQueue ( ) } -> updateOne ( [ '_id' => $ job [ '_id' ] , '$isolated' => true ] , [ '$setOnInsert' => $ job ] , [ 'upsert' => true ] ) ; if ( $ job [ 'options' ] [ 'at' ] <= $ now ) { $ this -> logger -> info ( 'postponed job [' . $ job [ '_id' ] . '] [' . $ job [ 'class' ] . '] can now be executed' , [ 'category' => get_class ( $ this ) , 'pm' => $ this -> process , ] ) ; unset ( $ this -> queue [ $ key ] ) ; $ job [ 'options' ] [ 'at' ] = 0 ; if ( true === $ this -> collectJob ( $ job , JobInterface :: STATUS_PROCESSING , JobInterface :: STATUS_POSTPONED ) ) { $ this -> processJob ( $ job ) ; } } } return true ; }
Check local queue for postponed jobs.
Check local queue for postponed jobs .