def run_forever(projection_stream, checkpoint, page_size, poll)
      position = checkpoint.latest
      loop do
        projection_stream.read_events(position, page_size, poll: poll) do |events|
          begin
            events.each do |event|
              route(event.name, event.metadata[:aggregate_id], **event.data)
              position = checkpoint.ack(position)
            end
          rescue RuntimeError => e
            puts e # TODO: Decide on a strategy.
            position = checkpoint.ack(position)
          end
        end
      end
    rescue RuntimeError => e
      puts e # TODO: Decide on a strategy.
      raise
    end
def run_forever(projection_stream, checkpoint, page_size, poll)
      position = checkpoint.latest
      loop do
        projection_stream.read_events(position, page_size, poll: poll) do |events|
          begin
            events.each do |event|
              route(event.name, event.metadata[:aggregate_id], **event.data)
              position = checkpoint.ack(position)
            end
          rescue RuntimeError => e
            puts e # TODO: Decide on a strategy.
            position = checkpoint.ack(position)
          end
        end
      end
    rescue RuntimeError => e
      puts e # TODO: Decide on a strategy.
      raise
    end
TODO: Make it stoppable.