protected function main(): void
    {
        $this->logger->info('start job listener', [
            'category' => get_class($this),
        ]);

        $cursor_jobs = $this->jobs->getCursor([
            '$or' => [
                ['status' => JobInterface::STATUS_WAITING],
                ['status' => JobInterface::STATUS_POSTPONED],
            ],
        ]);

        $cursor_events = $this->events->getCursor([
            'timestamp' => ['$gte' => new UTCDateTime()],
            'job' => ['$exists' => true],
            'status' => ['$gt' => JobInterface::STATUS_POSTPONED],
        ]);

        $this->catchSignal();

        while ($this->loop()) {
            while ($this->loop()) {
                if (null === $cursor_events->current()) {
                    if ($cursor_events->getInnerIterator()->isDead()) {
                        $this->logger->error('event queue cursor is dead, is it a capped collection?', [
                            'category' => get_class($this),
                        ]);

                        $this->events->create();

                        $this->main();

                        break;
                    }

                    $this->events->next($cursor_events, function () {
                        $this->main();
                    });
                }

                $event = $cursor_events->current();
                $this->events->next($cursor_events, function () {
                    $this->main();
                });

                if($event === null) {
                    break;
                }

                $this->handleEvent($event);
            }

            if (null === $cursor_jobs->current()) {
                if ($cursor_jobs->getInnerIterator()->isDead()) {
                    $this->logger->error('job queue cursor is dead, is it a capped collection?', [
                        'category' => get_class($this),
                    ]);

                    $this->jobs->create();
                    $this->main();

                    break;
                }

                $this->jobs->next($cursor_jobs, function () {
                    $this->main();
                });

                continue;
            }

            $job = $cursor_jobs->current();

            $this->jobs->next($cursor_jobs, function () {
                $this->main();
            });

            $this->handleJob($job);
        }
    }
protected function main(): void
    {
        $this->logger->info('start job listener', [
            'category' => get_class($this),
        ]);

        $cursor_jobs = $this->jobs->getCursor([
            '$or' => [
                ['status' => JobInterface::STATUS_WAITING],
                ['status' => JobInterface::STATUS_POSTPONED],
            ],
        ]);

        $cursor_events = $this->events->getCursor([
            'timestamp' => ['$gte' => new UTCDateTime()],
            'job' => ['$exists' => true],
            'status' => ['$gt' => JobInterface::STATUS_POSTPONED],
        ]);

        $this->catchSignal();

        while ($this->loop()) {
            while ($this->loop()) {
                if (null === $cursor_events->current()) {
                    if ($cursor_events->getInnerIterator()->isDead()) {
                        $this->logger->error('event queue cursor is dead, is it a capped collection?', [
                            'category' => get_class($this),
                        ]);

                        $this->events->create();

                        $this->main();

                        break;
                    }

                    $this->events->next($cursor_events, function () {
                        $this->main();
                    });
                }

                $event = $cursor_events->current();
                $this->events->next($cursor_events, function () {
                    $this->main();
                });

                if($event === null) {
                    break;
                }

                $this->handleEvent($event);
            }

            if (null === $cursor_jobs->current()) {
                if ($cursor_jobs->getInnerIterator()->isDead()) {
                    $this->logger->error('job queue cursor is dead, is it a capped collection?', [
                        'category' => get_class($this),
                    ]);

                    $this->jobs->create();
                    $this->main();

                    break;
                }

                $this->jobs->next($cursor_jobs, function () {
                    $this->main();
                });

                continue;
            }

            $job = $cursor_jobs->current();

            $this->jobs->next($cursor_jobs, function () {
                $this->main();
            });

            $this->handleJob($job);
        }
    }
protected function main ( ) : void { $ this -> logger -> info ( 'start job listener' , [ 'category' => get_class ( $ this ) , ] ) ; $ cursor_jobs = $ this -> jobs -> getCursor ( [ '$or' => [ [ 'status' => JobInterface :: STATUS_WAITING ] , [ 'status' => JobInterface :: STATUS_POSTPONED ] , ] , ] ) ; $ cursor_events = $ this -> events -> getCursor ( [ 'timestamp' => [ '$gte' => new UTCDateTime ( ) ] , 'job' => [ '$exists' => true ] , 'status' => [ '$gt' => JobInterface :: STATUS_POSTPONED ] , ] ) ; $ this -> catchSignal ( ) ; while ( $ this -> loop ( ) ) { while ( $ this -> loop ( ) ) { if ( null === $ cursor_events -> current ( ) ) { if ( $ cursor_events -> getInnerIterator ( ) -> isDead ( ) ) { $ this -> logger -> error ( 'event queue cursor is dead, is it a capped collection?' , [ 'category' => get_class ( $ this ) , ] ) ; $ this -> events -> create ( ) ; $ this -> main ( ) ; break ; } $ this -> events -> next ( $ cursor_events , function ( ) { $ this -> main ( ) ; } ) ; } $ event = $ cursor_events -> current ( ) ; $ this -> events -> next ( $ cursor_events , function ( ) { $ this -> main ( ) ; } ) ; if ( $ event === null ) { break ; } $ this -> handleEvent ( $ event ) ; } if ( null === $ cursor_jobs -> current ( ) ) { if ( $ cursor_jobs -> getInnerIterator ( ) -> isDead ( ) ) { $ this -> logger -> error ( 'job queue cursor is dead, is it a capped collection?' , [ 'category' => get_class ( $ this ) , ] ) ; $ this -> jobs -> create ( ) ; $ this -> main ( ) ; break ; } $ this -> jobs -> next ( $ cursor_jobs , function ( ) { $ this -> main ( ) ; } ) ; continue ; } $ job = $ cursor_jobs -> current ( ) ; $ this -> jobs -> next ( $ cursor_jobs , function ( ) { $ this -> main ( ) ; } ) ; $ this -> handleJob ( $ job ) ; } }
Fork handling, blocking process.
Fork handling blocking process .