这是一个解决方案,适用于在元素上Enumerator
使用 any 的任意数量的值:Ordering
import play.api.libs.iteratee._
import scala.concurrent._
object MergeEnums {
def apply[E: Ordering](enums: Enumerator[E]*)(implicit executor: ExecutionContext) = new Enumerator[E] {
def apply[A](iter: Iteratee[E, A]) = {
case class IterateeReturn(o: Option[(Promise[Promise[IterateeReturn]], E)])
val failP = Promise()
val failPF = failP.future
val initState = Future.traverse(enums) { enum =>
val p = Promise[IterateeReturn]()
enum.run(Iteratee.foldM(p) { (oldP: Promise[IterateeReturn], elem: E) =>
val p = Promise[Promise[IterateeReturn]]()
oldP success IterateeReturn(Some(p, elem))
p.future
} map { promise =>
promise success IterateeReturn(None)
}) onFailure { case t => failP failure t }
p.future
} map (_.map(_.o).flatten.toList)
Enumerator.unfoldM(initState) { fstate =>
Future.firstCompletedOf(Seq(fstate, failPF)) map { state =>
state.sortBy(_._2) match {
case Nil => None
case (oldP, elem) :: tail =>
val p = Promise[IterateeReturn]()
oldP success p
val newState = p.future.map(_.o.map(_ :: tail).getOrElse(tail))
Some(newState, elem)
}
}
} apply iter
}
}
}
它创建一个Iteratee
应用于Enumerator
传入的每个元素,并创建一个Enumerator
给排序的元素。Iteratee
实例和实例通过相互Enumerator
发送Promise
实例进行通信(因此是Promise[Promise[IterateeReturn]]
这样的)。