How to carry on executing Future sequence despite failure?

concurrency, future, scala

Solution

A genuinely useful thing (generally speaking) would be to be able to promote the error of a future into a proper value. Or in other words, transform a `Future[T]` into a `Future[Try[T]]` (the succesful return value becomes a `Success[T]` while the failure case becomes a `Failure[T]`). Here is how we might implement it:

// Can also be done more concisely (but less efficiently) as:
// f.map(Success(_)).recover{ case t: Throwable => Failure( t ) }
// NOTE: you might also want to move this into an enrichment class
def mapValue[T]( f: Future[T] ): Future[Try[T]] = {
  val prom = Promise[Try[T]]()
  f onComplete prom.success
  prom.future
}

Now, if you do the following:

Future.traverse(seq)( f andThen mapValue )

You'll obtain a succesful `Future[Seq[Try[A]]]`, whose eventual value contains a `Success` instance for each successful future, and a `Failure` instance for each failed future. If needed, you can then use `collect` on this seq to drop the `Failure` instances and keep only the sucessful values.

In other words, you can rewrite your helper method as follows:

def traverseFilteringErrors[A, B](seq: Seq[A])(f: A => Future[B]): Future[Seq[B]] = {
  Future.traverse( seq )( f andThen mapValue ) map ( _ collect{ case Success( x ) => x } )
}

Problem

The `traverse` method from `Future` object stops at first failure. I want a tolerant/forgiving version of this method which on occurrence of errors carries on with the rest of the sequence. Currently we have added the following method to our utils: ``` def traverseFilteringErrors[A, B <: AnyRef] (seq: Seq[A]) (f: A => Future[B]): Future[Seq[B]] = { val sentinelValue = null.asInstanceOf[B] val allResults = Future.traverse(seq) { x => f(x) recover { case _ => sentinelValue } } val successfulResults = allResults map { result => result.filterNot(_ == sentinelValue) } successfulResults } ``` Is there a better way to do this?

Original source