scala - sobre - series de netflix pdf

Scala: lista[futuro] a futuro[lista] sin tener en cuenta los futuros fallidos

Question

Estoy buscando una manera de convertir una lista de longitud arbitraria de futuros en un futuro de lista. Estoy usando Playframework, así que, en última instancia, lo que realmente quiero es un Future[Result] , pero para simplificar las cosas, digamos Future[List[Int]] La forma normal de hacer esto sería usar Future.sequence(...) pero hay un giro ... La lista que me da generalmente tiene alrededor de 10-20 futuros, y no es raro que uno de esos futuros falle (están haciendo solicitudes externas de servicio web). En lugar de tener que volver a intentarlos todos en el caso de que uno de ellos falle, me gustaría poder acceder a los que tuvieron éxito y devolverlos.

Por ejemplo, hacer lo siguiente no funciona

import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.Success
import scala.util.Failure

val listOfFutures = Future.successful(1) :: Future.failed(new Exception("Failure")) :: 
                    Future.successful(3) :: Nil

val futureOfList = Future.sequence(listOfFutures)

futureOfList onComplete {
  case Success(x) => println("Success!!! " + x)
  case Failure(ex) => println("Failed !!! " + ex)
}

scala> Failed !!! java.lang.Exception: Failure

En lugar de obtener la única excepción, me gustaría poder sacar el 1 y 3 de allí. Intenté usar Future.fold , pero aparentemente solo llama a Future.sequence detrás de escena.

Gracias de antemano por la ayuda!

Answer 1

Acabo de encontrar esta pregunta y tengo otra solución para ofrecer:

def allSuccessful[A, M[X] <: TraversableOnce[X]](in: M[Future[A]])
                                                (implicit cbf: CanBuildFrom[M[Future[A]], A, M[A]], 
                                                 executor: ExecutionContext): Future[M[A]] = {
    in.foldLeft(Future.successful(cbf(in))) {
      (fr, fa) ⇒ (for (r ← fr; a ← fa) yield r += a) fallbackTo fr
    } map (_.result())
}

La idea aquí es que dentro del doblez está esperando que se complete el siguiente elemento de la lista (usando la sintaxis de comprensión) y si el siguiente falla, simplemente recurra a lo que ya tiene.

Answer 2

El truco es primero asegurarse de que ninguno de los futuros haya fallado. .recover es tu amigo aquí, puedes combinarlo con un map para convertir todos los resultados de Future[T] a Future[Try[T]]] , todos los cuales tendrán un futuro exitoso.

Nota: Puede usar Option o Either también aquí, pero Try es la manera más limpia si específicamente desea atrapar excepciones.

def futureToFutureTry[T](f: Future[T]): Future[Try[T]] =
  f.map(Success(_)).recover(x => Failure(x))

val listOfFutures = ...
val listOfFutureTrys = listOfFutures.map(futureToFutureTry(_))

Luego use Future.sequence como antes, para darle un Future[List[Try[T]]]

val futureListOfTrys = Future.sequence(listOfFutureTrys)

Luego filtra:

val futureListOfSuccesses = futureListOfTrys.map(_.filter(_.isSuccess))

Incluso puede extraer los fallos específicos, si los necesita:

val futureListOfFailures = futureListOfTrys.map(_.filter(_.isFailure))

Answer 3

Probé la respuesta de Kevin, y encontré un error en mi versión de Scala (2.11.5) ... Corregí eso, y escribí algunas pruebas adicionales si alguien está interesado ... aquí está mi versión>

implicit class FutureCompanionOps(val f: Future.type) extends AnyVal {

    /** Given a list of futures `fs`, returns the future holding the list of Try''s of the futures from `fs`.
      * The returned future is completed only once all of the futures in `fs` have been completed.
      */
    def allAsTrys[T](fItems: /* future items */ List[Future[T]]): Future[List[Try[T]]] = {
      val listOfFutureTrys: List[Future[Try[T]]] = fItems.map(futureToFutureTry)
      Future.sequence(listOfFutureTrys)
    }

    def futureToFutureTry[T](f: Future[T]): Future[Try[T]] = {
      f.map(Success(_)) .recover({case x => Failure(x)})
    }

    def allFailedAsTrys[T](fItems: /* future items */ List[Future[T]]): Future[List[Try[T]]] = {
      allAsTrys(fItems).map(_.filter(_.isFailure))
    }

    def allSucceededAsTrys[T](fItems: /* future items */ List[Future[T]]): Future[List[Try[T]]] = {
      allAsTrys(fItems).map(_.filter(_.isSuccess))
    }
}


// Tests... 



  // allAsTrys tests
  //
  test("futureToFutureTry returns Success if no exception") {
    val future =  Future.futureToFutureTry(Future{"mouse"})
    Thread.sleep(0, 100)
    val futureValue = future.value
    assert(futureValue == Some(Success(Success("mouse"))))
  }
  test("futureToFutureTry returns Failure if exception thrown") {
    val future =  Future.futureToFutureTry(Future{throw new IllegalStateException("bad news")})
    Thread.sleep(5)            // need to sleep a LOT longer to get Exception from failure case... interesting.....
    val futureValue = future.value

    assertResult(true) {
      futureValue match {
        case Some(Success(Failure(error: IllegalStateException)))  => true
      }
    }
  }
  test("Future.allAsTrys returns Nil given Nil list as input") {
    val future =  Future.allAsTrys(Nil)
    assert ( Await.result(future, 100 nanosecond).isEmpty )
  }
  test("Future.allAsTrys returns successful item even if preceded by failing item") {
    val future1 =  Future{throw new IllegalStateException("bad news")}
    var future2 = Future{"dog"}

    val futureListOfTrys =  Future.allAsTrys(List(future1,future2))
    val listOfTrys =  Await.result(futureListOfTrys, 10 milli)
    System.out.println("successItem:" + listOfTrys);

    assert(listOfTrys(0).failed.get.getMessage.contains("bad news"))
    assert(listOfTrys(1) == Success("dog"))
  }
  test("Future.allAsTrys returns successful item even if followed by failing item") {
    var future1 = Future{"dog"}
    val future2 =  Future{throw new IllegalStateException("bad news")}

    val futureListOfTrys =  Future.allAsTrys(List(future1,future2))
    val listOfTrys =  Await.result(futureListOfTrys,  10 milli)
    System.out.println("successItem:" + listOfTrys);

    assert(listOfTrys(1).failed.get.getMessage.contains("bad news"))
    assert(listOfTrys(0) == Success("dog"))
  }
  test("Future.allFailedAsTrys returns the failed item and only that item") {
    var future1 = Future{"dog"}
    val future2 =  Future{throw new IllegalStateException("bad news")}

    val futureListOfTrys =  Future.allFailedAsTrys(List(future1,future2))
    val listOfTrys =  Await.result(futureListOfTrys,  10 milli)
    assert(listOfTrys(0).failed.get.getMessage.contains("bad news"))
    assert(listOfTrys.size == 1)
  }
  test("Future.allSucceededAsTrys returns the succeeded item and only that item") {
    var future1 = Future{"dog"}
    val future2 =  Future{throw new IllegalStateException("bad news")}

    val futureListOfTrys =  Future.allSucceededAsTrys(List(future1,future2))
    val listOfTrys =  Await.result(futureListOfTrys,  10 milli)
    assert(listOfTrys(0) == Success("dog"))
    assert(listOfTrys.size == 1)
  }

Answer 4

Puede ajustar fácilmente el resultado futuro con la opción y luego aplanar la lista:

def futureToFutureOption[T](f: Future[T]): Future[Option[T]] =
    f.map(Some(_)).recover {
      case e => None
    }
val listOfFutureOptions = listOfFutures.map(futureToFutureOption(_))

val futureListOfOptions = Future.sequence(listOfFutureOptions)

val futureListOfSuccesses = futureListOfOptions.flatten

Answer 5

Scala 2.12 tiene una mejora en Future.transform que se presta al analizador con menos códigos.

val futures = Seq(Future{1},Future{throw new Exception})

val seq = Future.sequence(futures.map(_.transform(Success(_)))) // instead of map and recover

@val successes = seq.map(_.collect{case Success(x)=>x})
successes: Future[Seq[Int]] = Future(Success(List(1)))

@val failures = seq.map(_.collect{case Failure(x)=>x})
failures: Future[Seq[Throwable]] = Future(Success(List(java.lang.Exception)))