public function listen(Closure $callback, array $query = []): self
    {
        if (0 === count($query)) {
            $query = [
                'timestamp' => ['$gte' => new UTCDateTime()],
            ];
        }

        $cursor = $this->events->getCursor($query);

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

                    $this->events->create();

                    return $this->listen($callback, $query);
                }

                $this->events->next($cursor, function () use ($callback, $query) {
                    return $this->listen($callback, $query);
                });

                continue;
            }

            $result = $cursor->current();
            $this->events->next($cursor, function () use ($callback, $query) {
                $this->listen($callback, $query);
            });

            $process = new Process($result, $this, $this->events);
            if (true === $callback($process)) {
                return $this;
            }
        }
    }
public function listen(Closure $callback, array $query = []): self
    {
        if (0 === count($query)) {
            $query = [
                'timestamp' => ['$gte' => new UTCDateTime()],
            ];
        }

        $cursor = $this->events->getCursor($query);

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

                    $this->events->create();

                    return $this->listen($callback, $query);
                }

                $this->events->next($cursor, function () use ($callback, $query) {
                    return $this->listen($callback, $query);
                });

                continue;
            }

            $result = $cursor->current();
            $this->events->next($cursor, function () use ($callback, $query) {
                $this->listen($callback, $query);
            });

            $process = new Process($result, $this, $this->events);
            if (true === $callback($process)) {
                return $this;
            }
        }
    }
public function listen ( Closure $ callback , array $ query = [ ] ) : self { if ( 0 === count ( $ query ) ) { $ query = [ 'timestamp' => [ '$gte' => new UTCDateTime ( ) ] , ] ; } $ cursor = $ this -> events -> getCursor ( $ query ) ; while ( true ) { if ( null === $ cursor -> current ( ) ) { if ( $ cursor -> getInnerIterator ( ) -> isDead ( ) ) { $ this -> logger -> error ( 'events queue cursor is dead, is it a capped collection?' , [ 'category' => get_class ( $ this ) , ] ) ; $ this -> events -> create ( ) ; return $ this -> listen ( $ callback , $ query ) ; } $ this -> events -> next ( $ cursor , function ( ) use ( $ callback , $ query ) { return $ this -> listen ( $ callback , $ query ) ; } ) ; continue ; } $ result = $ cursor -> current ( ) ; $ this -> events -> next ( $ cursor , function ( ) use ( $ callback , $ query ) { $ this -> listen ( $ callback , $ query ) ; } ) ; $ process = new Process ( $ result , $ this , $ this -> events ) ; if ( true === $ callback ( $ process ) ) { return $ this ; } } }
Listen for events.
Listen for events .