trasformare un Seq [Futuro [X]] in un Enumeratore [X]

Question

C'è un modo per trasformare un Seq [Futuro [X]] in un Enumerator[X]? Il caso d'uso è che voglio ottenere risorse eseguendo la scansione del Web. Questo restituirà una Sequenza di Futures, e mi piacerebbe restituire un Enumeratore che spingerà i futuri nell'ordine in cui sono finiti per la prima volta sull'Iteratee.

Sembra che lo standard Future select gist di Victor Klang possa essere utilizzato per farlo, anche se sembra piuttosto inefficiente.

Nota: I Iteratees e Enumeratore di in questione sono quelle indicate dalla versione 2.x quadro di gioco, vale a dire con le seguenti importazioni: import play.api.libs.iteratee._

Answer 1

A, migliore risposta più breve e penso che più efficiente è:

 
    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))) 
      } 
     } 
     } 
    } 

Answer 2

Utilizzando 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) 
    } 
} 

posso quindi ottenere quello che ho necessario con

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

Answer 3

È possibile costruirne uno utilizzando il servizio di esecuzione di Java Executor (JavaDoc). L'idea è di creare una sequenza di nuovi futures, ognuno utilizzando ExecutorCompletionService.take() per attendere il risultato successivo. Ogni futuro inizierà, quando il futuro precedente avrà il suo risultato.

Ma per favore sappiate che questo potrebbe non essere così efficiente, perché molta sincronizzazione sta accadendo dietro le quinte. Potrebbe essere più efficiente utilizzare una riduzione della mappa parallela per il calcolo (ad esempio utilizzando ParSeq di Scala) e lasciare che l'Enumeratore attenda il risultato completo.

Answer 4

ATTENZIONE: Non compilato prima di rispondere

Che dire qualcosa di simile:

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)))) } 
} 

Answer 5

Qui è qualcosa che ho trovato a portata di mano,

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)) 
} 

Answer 6

Mi rendo conto che la questione è un po 'già vecchio, ma in base alla risposta di Santhosh e il built-in Enumterator.enumerate() implementazione mi si avvicinò con il seguente:

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 
     } 
    } 
    } 
} 

Si noti che a differenza del primo Viktor- select-based-solution questo conserva l'ordine, ma è ancora possibile avviare tutti i calcoli in modo asincrono prima. Così, per esempio, è possibile effettuare le seguenti operazioni:

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

Quest'ultimo metodo era in realtà quello che cercavo quando mi sono imbattuto in questo thread. Spero che aiuti qualcuno! :)

Answer 7

Vorrei proporre l'uso di una trasmissione

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 
    } 

ho aggiunto ErrorHandling e defaultValues ​​ma si può saltare quelli utilizzando onSuccess o onFailure, invece di onComplete