scala - 将 Seq[Future[X]] 转换为 Enumerator[X]

Question

有没有办法将 Seq[Future[X]] 变成 Enumerator[X] ?用例是我想通过爬网来获取资源。这将返回一个 Future 序列，我想返回一个 Enumerator，它将按照它们首先完成的顺序推送到 Iteratee。

它看起来像 Victor Klang 的 Future select gist可以用来做到这一点 - 尽管它看起来效率很低。

注意:有问题的迭代器和枚举器是由 play 框架版本 2.x 给出的，即具有以下导入:import play.api.libs.iteratee._

Answer 1

使用 Victor Klang's select method :

  /**
   * "Select" off the first future to be satisfied.  Return this as a
   * result, with the remainder of the Futures as a sequence.
   *
   * @param fs a scala.collection.Seq
   */
  def select[A](fs: Seq[Future[A]])(implicit ec: ExecutionContext): 
      Future[(Try[A], Seq[Future[A]])] = {
    @scala.annotation.tailrec
    def stripe(p: Promise[(Try[A], Seq[Future[A]])],
               heads: Seq[Future[A]],
               elem: Future[A],
               tail: Seq[Future[A]]): Future[(Try[A], Seq[Future[A]])] = {
      elem onComplete { res => if (!p.isCompleted) p.trySuccess((res, heads ++ tail)) }
      if (tail.isEmpty) p.future
      else stripe(p, heads :+ elem, tail.head, tail.tail)
    }
    if (fs.isEmpty) Future.failed(new IllegalArgumentException("empty future list!"))
    else stripe(Promise(), fs.genericBuilder[Future[A]].result, fs.head, fs.tail)
   }
}

然后我可以得到我需要的东西


    Enumerator.unfoldM(initialSeqOfFutureAs){ seqOfFutureAs =>
        if (seqOfFutureAs.isEmpty) {
          Future(None)
        } else {
          FutureUtil.select(seqOfFutureAs).map {
            case (t, seqFuture) => t.toOption.map {
              a => (seqFuture, a)
            }
          }
        }
    }