Futures / Success race

by

The use case is the first successful completion:

scala> :pa
// Entering paste mode (ctrl-D to finish)

def firstSuccessOf[T](fs: Future[T]*)(implicit x: ExecutionContext): Future[T] = {
  val p = Promise[T]()
  val count = new java.util.concurrent.atomic.AtomicInteger(fs.size)
  def bad() = if (count.decrementAndGet == 0) { p tryComplete new Failure(new RuntimeException("All bad")) }
  val completeFirst: Try[T] => Unit = p tryComplete _
  fs foreach { _ onComplete { case v @ Success(_) => completeFirst(v) case _ => bad() }}
  p.future
}

// Exiting paste mode, now interpreting.

firstSuccessOf: [T](fs: scala.concurrent.Future[T]*)(implicit x: scala.concurrent.ExecutionContext)scala.concurrent.Future[T]

so

scala> def f = Future { Thread sleep 5000L ; println("Failing") ; throw new NullPointerException }
f: scala.concurrent.Future[Nothing]

scala> def g = Future { Thread sleep 10000L ; println("OK") ; 7 }
g: scala.concurrent.Future[Int]

scala> firstSuccessOf(f,g)
res3: scala.concurrent.Future[Int] = scala.concurrent.impl.Promise$DefaultPromise@5ed53f6b

scala> res0Failing
          3.value
res4: Option[scala.util.Try[Int]] = None

scala> res3.valueOK

res5: Option[scala.util.Try[Int]] = Some(Success(7))

or

scala> def h = Future { Thread sleep 7000L ; println("Failing too") ; throw new NullPointerException }
h: scala.concurrent.Future[Nothing]


scala> firstSuccessOf(f,h)
res10: scala.concurrent.Future[Nothing] = scala.concurrent.impl.Promise$DefaultPromise@318d30be

scala> 

scala> res10.Failing
value
res11: Option[scala.util.Try[Nothing]] = None

scala> Failing too


scala> res10.value
res12: Option[scala.util.Try[Nothing]] = Some(Failure(java.lang.RuntimeException: All bad))

@ ysusuk ‘s answer is what Future.firstCompletedOf does under the hood.

More Related Contents:

  1. How to wait for several Futures?
  2. Futures do not run before program termination
  3. Scala Futures – built in timeout?
  4. Scala: List[Future] to Future[List] disregarding failed futures
  5. Scala’s “for comprehension” with futures
  6. Accessing value returned by scala futures
  7. How do I make a function involving futures tail recursive?
  8. Scala recover or recoverWith
  9. Can Scala actors process multiple messages simultaneously?
  10. error in if return in Scala
  11. How do I get around type erasure on Scala? Or, why can’t I get the type parameter of my collections?
  12. Call by name vs call by value in Scala, clarification needed
  13. What does the `#` operator mean in Scala?
  14. How do I skip a header from CSV files in Spark?
  15. What is meant by Scala’s path-dependent types?
  16. how to keep return value when logging in scala
  17. Finding type parameters via reflection in Scala 2.10?
  18. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  19. withFilter instead of filter
  20. Scala multiple type pattern matching
  21. What are the rules to govern underscore to define anonymous function?
  22. Use case of scala.concurrent.blocking
  23. Scala Macros: Making a Map out of fields of a class in Scala
  24. Apache Spark: Get number of records per partition
  25. Derive multiple columns from a single column in a Spark DataFrame
  26. Converting nested case classes to nested Maps using Shapeless
  27. Functional Reactive Programming in Scala [closed]
  28. What is a Singleton Type exactly?
  29. How to create a Dataset of Maps?
  30. Spark ML VectorAssembler returns strange output

Leave a Comment