public function addJobOnce(string $class, $data, array $options = []): Process
    {
        $filter = [
            'class' => $class,
            '$or' => [
                ['status' => JobInterface::STATUS_WAITING],
                ['status' => JobInterface::STATUS_POSTPONED],
                ['status' => JobInterface::STATUS_PROCESSING],
            ],
        ];

        $requested = $options;
        $document = $this->prepareInsert($class, $data, $options);

        if (true !== $options[self::OPTION_IGNORE_DATA]) {
            $filter = ['data' => $data] + $filter;
        }

        $result = $this->db->{$this->job_queue}->updateOne($filter, ['$setOnInsert' => $document], [
            'upsert' => true,
            '$isolated' => true,
        ]);

        if ($result->getMatchedCount() > 0) {
            $document = $this->db->{$this->job_queue}->findOne($filter, [
                'typeMap' => self::TYPE_MAP,
            ]);

            if (array_intersect_key($document['options'], $requested) !== $requested || ($data !== $document['data'] && true === $options[self::OPTION_IGNORE_DATA])) {
                $this->logger->debug('job ['.$document['_id'].'] options/data changed, reschedule new job', [
                    'category' => get_class($this),
                    'data' => $data,
                ]);

                $this->cancelJob($document['_id']);

                return $this->addJobOnce($class, $data, $options);
            }

            return new Process($document, $this, $this->events);
        }

        $this->logger->debug('queue job ['.$result->getUpsertedId().'] added to ['.$class.']', [
            'category' => get_class($this),
            'params' => $options,
            'data' => $data,
        ]);

        $this->db->{$this->event_queue}->insertOne([
            'job' => $result->getUpsertedId(),
            'status' => JobInterface::STATUS_WAITING,
            'timestamp' => new UTCDateTime(),
        ]);

        $document = $this->db->{$this->job_queue}->findOne(['_id' => $result->getUpsertedId()], [
            'typeMap' => self::TYPE_MAP,
        ]);

        return new Process($document, $this, $this->events);
    }
public function addJobOnce(string $class, $data, array $options = []): Process
    {
        $filter = [
            'class' => $class,
            '$or' => [
                ['status' => JobInterface::STATUS_WAITING],
                ['status' => JobInterface::STATUS_POSTPONED],
                ['status' => JobInterface::STATUS_PROCESSING],
            ],
        ];

        $requested = $options;
        $document = $this->prepareInsert($class, $data, $options);

        if (true !== $options[self::OPTION_IGNORE_DATA]) {
            $filter = ['data' => $data] + $filter;
        }

        $result = $this->db->{$this->job_queue}->updateOne($filter, ['$setOnInsert' => $document], [
            'upsert' => true,
            '$isolated' => true,
        ]);

        if ($result->getMatchedCount() > 0) {
            $document = $this->db->{$this->job_queue}->findOne($filter, [
                'typeMap' => self::TYPE_MAP,
            ]);

            if (array_intersect_key($document['options'], $requested) !== $requested || ($data !== $document['data'] && true === $options[self::OPTION_IGNORE_DATA])) {
                $this->logger->debug('job ['.$document['_id'].'] options/data changed, reschedule new job', [
                    'category' => get_class($this),
                    'data' => $data,
                ]);

                $this->cancelJob($document['_id']);

                return $this->addJobOnce($class, $data, $options);
            }

            return new Process($document, $this, $this->events);
        }

        $this->logger->debug('queue job ['.$result->getUpsertedId().'] added to ['.$class.']', [
            'category' => get_class($this),
            'params' => $options,
            'data' => $data,
        ]);

        $this->db->{$this->event_queue}->insertOne([
            'job' => $result->getUpsertedId(),
            'status' => JobInterface::STATUS_WAITING,
            'timestamp' => new UTCDateTime(),
        ]);

        $document = $this->db->{$this->job_queue}->findOne(['_id' => $result->getUpsertedId()], [
            'typeMap' => self::TYPE_MAP,
        ]);

        return new Process($document, $this, $this->events);
    }
public function addJobOnce ( string $ class , $ data , array $ options = [ ] ) : Process { $ filter = [ 'class' => $ class , '$or' => [ [ 'status' => JobInterface :: STATUS_WAITING ] , [ 'status' => JobInterface :: STATUS_POSTPONED ] , [ 'status' => JobInterface :: STATUS_PROCESSING ] , ] , ] ; $ requested = $ options ; $ document = $ this -> prepareInsert ( $ class , $ data , $ options ) ; if ( true !== $ options [ self :: OPTION_IGNORE_DATA ] ) { $ filter = [ 'data' => $ data ] + $ filter ; } $ result = $ this -> db -> { $ this -> job_queue } -> updateOne ( $ filter , [ '$setOnInsert' => $ document ] , [ 'upsert' => true , '$isolated' => true , ] ) ; if ( $ result -> getMatchedCount ( ) > 0 ) { $ document = $ this -> db -> { $ this -> job_queue } -> findOne ( $ filter , [ 'typeMap' => self :: TYPE_MAP , ] ) ; if ( array_intersect_key ( $ document [ 'options' ] , $ requested ) !== $ requested || ( $ data !== $ document [ 'data' ] && true === $ options [ self :: OPTION_IGNORE_DATA ] ) ) { $ this -> logger -> debug ( 'job [' . $ document [ '_id' ] . '] options/data changed, reschedule new job' , [ 'category' => get_class ( $ this ) , 'data' => $ data , ] ) ; $ this -> cancelJob ( $ document [ '_id' ] ) ; return $ this -> addJobOnce ( $ class , $ data , $ options ) ; } return new Process ( $ document , $ this , $ this -> events ) ; } $ this -> logger -> debug ( 'queue job [' . $ result -> getUpsertedId ( ) . '] added to [' . $ class . ']' , [ 'category' => get_class ( $ this ) , 'params' => $ options , 'data' => $ data , ] ) ; $ this -> db -> { $ this -> event_queue } -> insertOne ( [ 'job' => $ result -> getUpsertedId ( ) , 'status' => JobInterface :: STATUS_WAITING , 'timestamp' => new UTCDateTime ( ) , ] ) ; $ document = $ this -> db -> { $ this -> job_queue } -> findOne ( [ '_id' => $ result -> getUpsertedId ( ) ] , [ 'typeMap' => self :: TYPE_MAP , ] ) ; return new Process ( $ document , $ this , $ this -> events ) ; }
Only add job if not in queue yet.
Only add job if not in queue yet .