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;
            }
        }
    }