通过编写自己的GraphStage找到了解决方案

final class PaginationGraphStage[S <: Uri, E](f: S => Future[Option[(S, E)]])(implicit ec: ExecutionContextExecutor)
  extends GraphStage[FlowShape[S, E]]{
  val in: Inlet[S] = Inlet[S]("PaginationGraphStage.in")
  val out: Outlet[E] = Outlet[E]("PaginationGraphStage.out")
  override val shape: FlowShape[S, E] = FlowShape.of(in, out)
  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) with OutHandler with InHandler {
      private[this] var state: S = _
      private[this] var inFlight = 0
      private[this] var asyncFinished = false
      private[this] def todo: Int = inFlight
      def futureCompleted(result: Try[Option[(Uri, E)]]): Unit = {
        inFlight -= 1
        result match {
          case Failure(ex) => fail(out, ex)
          case Success(None) =>
            asyncFinished = true
            complete(out)
          case Success(Some((newS: S, elem: E))) if !newS.isEmpty =>
            push(out, elem)
            state = newS
          case Success(Some((newS: Uri, elem: E))) =>
            push(out, elem)
            asyncFinished = true
            if (isAvailable(in)) getHandler(in).onPush()
            else completeStage()
        }
      }
      private val futureCB = getAsyncCallback(futureCompleted)
      private val invokeFutureCB: Try[Option[(S, E)]] => Unit = futureCB.invoke
      private def pullIfNeeded(): Unit = {
        if (!hasBeenPulled(in)) tryPull(in)
      }
      override def onUpstreamFinish(): Unit = {
        if (todo == 0) completeStage()
      }
      def onPull(): Unit = {
        if (state != null) {
          asyncFinished = false
          inFlight += 1
          val future = f(state)
          future.value match {
            case None => future.onComplete(invokeFutureCB)
            case Some(v) => futureCompleted(v)
          }
        } else {
          pullIfNeeded()
        }
      }
      override def onPush(): Unit = {
        if (state == null) {
          inFlight += 1
          state = grab(in)
          pullIfNeeded()
          getHandler(out).onPull()
        }
        if (asyncFinished) {
          inFlight += 1
          state = grab(in)
          pullIfNeeded()
        }
      }
      setHandlers(in, out, this)
    }
}

字符串

Source.unfoldAsync就是你要找的。
我准备了一个简单的项目，它遍历REST API的所有页面，并累积所有页面的结果，返回Future with Seq。
您可以在over on GitHub中找到完整的源代码和项目

class CatsHttpClientImpl(implicit system: ActorSystem[_], ec: ExecutionContext) extends CatsHttpClient {
  private val logger: Logger = LoggerFactory.getLogger(classOf[CatsHttpClientImpl])
  private val start: Option[String] = Some("https://catfact.ninja/breeds")
  override def getAllBreads: Future[Seq[Cat]] = {
    Source
      .unfoldAsync(start) {
        case Some(next) =>
          val nextChunkFuture: Future[CatsResponse] = sendRequest(next)
          nextChunkFuture.map { resp =>
            resp.nextPageUrl match {
              case Some(url) => Some((Some(url), resp.data))
              case None => Some((None, resp.data))
            }
          }
        case None => Future.successful(None)
      }
      .runWith(Sink.fold(Seq(): Seq[Cat])(_ ++ _))
  }
  private def sendRequest(url: String): Future[CatsResponse] = {
    logger.info(s"CatsHttpClientImpl: Sending request $url")
    val request = HttpRequest(
      uri = Uri(url),
      headers = List(
        RawHeader("Accept", "application/json")
      )
    )
    Http(system).singleRequest(request).flatMap { response =>
      response.status match {
        case StatusCodes.OK =>
          logger.info("CatsHttpClientImpl: Received success")
          Unmarshal(response.entity).to[CatsResponse]
        case _ =>
          logger.error("CatsHttpClientImpl: Received error")
          throw new CatsHttpClientException()
      }
    }
  }
}

字符串