Преобразование Seq [Future [X]] в Enumerator [X]

Ответ 1

Лучше, короче, и я думаю, что более эффективный ответ:

   def toEnumerator(seqFutureX: Seq[Future[X]]) = new Enumerator[X] { 
      def apply[A](i: Iteratee[X, A]): Future[Iteratee[X, A]] = {
        Future.sequence(seqFutureX).flatMap { seqX: Seq[X] => 
            seqX.foldLeft(Future.successful(i)) {
              case (i, x) => i.flatMap(_.feed(Input.El(x)))
            }
        }
      }
    }

Ответ 2

Использование Метод выбора Виктора Клана:

  /**
   * "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)
            }
          }
        }
    }

Ответ 3

Я понимаю, что вопрос уже немного старый, но на основе ответа Santhosh и встроенной реализации Enumterator.enumerate() я пришел к следующему:

def enumerateM[E](traversable: TraversableOnce[Future[E]])(implicit ec: ExecutionContext): Enumerator[E] = {
  val it = traversable.toIterator
  Enumerator.generateM {
    if (it.hasNext) {
      val next: Future[E] = it.next()
      next map {
        e => Some(e)
      }
    } else {
      Future.successful[Option[E]] {
        None
      }
    }
  }
}

Обратите внимание, что в отличие от первого решения, основанного на выборе Виктора, это сохраняет порядок, но вы все равно можете начинать все вычисления асинхронно раньше. Так, например, вы можете сделать следующее:

// For lack of a better name
def mapEachM[E, NE](eventuallyList: Future[List[E]])(f: E => Future[NE])(implicit ec: ExecutionContext): Enumerator[NE] =
  Enumerator.flatten(
    eventuallyList map { list =>
      enumerateM(list map f)
    }
  )

Этот последний метод был тем, что я искал, когда наткнулся на эту тему. Надеюсь, это поможет кому-то!:)

Ответ 4

Вы можете создать его с помощью службы Java Executor Completeion (JavaDoc). Идея состоит в том, чтобы создать последовательность новых фьючерсов, каждая из которых использует ExecutorCompletionService.take() для ожидания следующего результата. Каждое будущее начнется, когда предыдущее будущее имеет свой результат.

Но, пожалуйста, знайте, что это может быть не так эффективно, потому что много синхронизации происходит за кулисами. Возможно, было бы более эффективно использовать некоторое сокращение параллельной карты для вычисления (например, с помощью Scala ParSeq) и позволить Enumerator ждать полного результата.

Ответ 5

ПРЕДУПРЕЖДЕНИЕ: не скомпилировано перед ответом

Что-то вроде этого:

def toEnumerator(seqFutureX: Seq[Future[X]]) = new Enumerator[X] { 
  def apply[A](i: Iteratee[X, A]): Future[Iteratee[X, A]] = 
    Future.fold(seqFutureX)(i){ case (i, x) => i.flatMap(_.feed(Input.El(x)))) }
}

Ответ 6

Вот что я нашел удобным,

def unfold[A,B](xs:Seq[A])(proc:A => Future[B])(implicit errorHandler:Throwable => B):Enumerator[B] = {
    Enumerator.unfoldM (xs) { xs =>
        if (xs.isEmpty) Future(None)
        else proc(xs.head) map (b => Some(xs.tail,b)) recover {
            case e => Some((xs.tail,errorHandler(e)))
        }
    }
}

def unfold[A,B](fxs:Future[Seq[A]])(proc:A => Future[B]) (implicit errorHandler1:Throwable => Seq[A], errorHandler:Throwable => B) :Enumerator[B] = {

    (unfold(Seq(fxs))(fxs => fxs)(errorHandler1)).flatMap(unfold(_)(proc)(errorHandler))
}

def unfoldFutures[A,B](xsfxs:Seq[Future[Seq[A]]])(proc:A => Future[B]) (implicit errorHandler1:Throwable => Seq[A], errorHandler:Throwable => B) :Enumerator[B] = {

    xsfxs.map(unfold(_)(proc)).reduceLeft((a,b) => a.andThen(b))
}

Ответ 7

Я хотел бы предложить использовать широковещательную рассылку

def seqToEnumerator[A](futuresA: Seq[Future[A]])(defaultValue: A, errorHandler: Throwable => A): Enumerator[A] ={
    val (enumerator, channel) = Concurrent.broadcast[A]
    futuresA.foreach(f => f.onComplete({
      case Success(Some(a: A)) => channel.push(a)
      case Success(None) => channel.push(defaultValue)
      case Failure(exception) => channel.push(errorHandler(exception))
    }))
    enumerator
  }

Я добавил errorHandling и defaultValues, но вы можете пропустить их, используя onSuccess или onFailure, а не onComplete