Future Combinators

Published 2014-09-22. Last modified 2019-07-01.
Time to read: 12 minutes.

This lecture consists of a discussion and a code example for every important method provided by the Scala Future class and companion object. This lecture is the culmination of a long sequence of information that started with the Introduction to Scala course and cross-references to other lectures are provided to help you understand the code examples.

You may remember from the Combinators lecture earlier in this course that monads hold values, and Future is just a regular monad like all others. Performing an operation on a monad yields another monad of the same type.

That same lecture taught you that combinators are methods that perform transformations on the contents of monads. Futures are composable, which means:

Operations on a Future or a collection of Futures can be chained together without blocking.
Transformations such as map and flatMap can be applied to collections of Futures.
Collections of Futures can be manipulated (for example, reduced).

Futures are immutable, so the result of transforming a Future with a combinator is a new Future. Chaining multiple combinators together results in intermediate Futures being created. Some combinators accept partial functions; as you know from the Partial Functions lecture, partial functions which each handle a subset of possible input values or types can be chained together, and the resulting chain of partial functions is able to handle the union of all the input types and values of the individual partial functions.

When many combinators are applied to Futures, however the cost is that a new intermediate Future will be created for each combinator, and that means more expensive context switching. When the partial functions are computationally expensive, the extra overhead of the additional context switches is not noticeable, however a chain of lightweight partial functions would be more efficient if they were consolidated into one partial function with many case statements. Examples of this are shown where appropriate in this lecture.

The sample code for this lecture can be found in courseNotes/src/main/scala/multi/futures/FutureComb.scala.

Many of the methods shown require an ExecutionContext; the code shown on this page assumes that an ExecutionContext is provided. As we discussed in the MultiThreading lecture, one way to provide an ExecutionContext is to place the following before the methods that require it:

Scala code

import concurrent.ExecutionContext.Implicits.global

Some of the code example use postfix notation. You can enable postfix notation with the following:

Scala code

import language.postfixOps

Exceptions encountered in this lecture can be imported as follows.

Scala code

import java.io.{FileNotFoundException, IOException}
import java.util.concurrent.TimeoutException
import java.net.{MalformedURLException, UnknownHostException}

Some of the code examples require random numbers. This random number generator was used:

Scala code

val random = new util.Random()

Combinators for Handling Failure

This section introduces the failure-related combinators. The code examples in this section rely on the following variable values:

Scala code

lazy val goodUrlStr1       = "https://www.scalacourses.com"
lazy val goodUrlStr2       = "https://www.micronauticsresearch.com"
lazy val badHostUrlStr     = "https://www.not-really-here.com"
lazy val badPageUrlStr     = "https://scalacourses.com/noPageHere"
lazy val badProtocolUrlStr = "blah://scalacourses.com"

lazy val badHostFuture: Future[String]     = readUrlFuture(badHostUrlStr)
lazy val badPageFuture: Future[String]     = readUrlFuture(badHostUrlStr)
lazy val badProtocolFuture: Future[String] = readUrlFuture(badHostUrlStr)
lazy val defaultFuture: Future[String]     = readUrlFuture(goodUrlStr1)

readUrlFuture is a method defined in the multi package object. Package objects were introduced in the Scala Imports and Packages lecture of the Introduction to Scala course. This asynchronous method returns a Future of the contents of the web page pointed to by the urlStr parameter. The method is said to be asynchronous because it returns a Future. If the web page is longer than the optional maxChars parameter, it is truncated and ellipses are appended to indicate truncation.

Scala code

/** @return up to first maxChars characters of web page at given url */
def readUrl(url: String, maxChars: Int=500): String = {
  val contents = io.Source.fromURL(url).mkString.trim
  contents.substring(0, math.min(contents.length, maxChars)) +
    (if (contents.length>maxChars) "..." else "")
}

/** @return Future of first maxChars characters of web page at given url */
def readUrlFuture(urlStr: String, maxChars: Int=500): Future[String] =
  Future(readUrl(urlStr, maxChars))

Some of the code examples also use the show method. Take a moment to study it; you should be able to understand what it does by this point in the course.

Scala code

/** Blocks until contents of web page at urlStr or whatever recoveryFn contains becomes available, then prints contents.
 * @param urlStr String passed to java.net.URL that specifies the URL of a web page
 * @param msg Optional parameter that specifies a prefix message to be displayed before the web page contents
 * @param recoveryFn Function1[Future[String], Future[String]] which is a pluggable recovery strategy for failed web pages
 * @return Future of contents of web page at urlStr or recoverFn */
def show(urlStr: String, msg: String="")
        (recoveryFn: Future[String] => Future[String]): Future[String] = {
  val future = recoveryFn(readUrlFuture(urlStr))
  println(s"$urlStr; ${ if (msg.length>0) s"$msg, " else "" }returning " +
    Await.result(future, 30 minutes))
  future
}

The show method’s recoveryFn parameter has type Function1[Future[String] => Future[String]], which is a recovery strategy for failed web pages. If this parameter's type is not familiar to you, please review the Functions are First Class and Lambda Review & Drill lectures of the Introduction to Scala course. This method allows us to write some succinct and efficient code that is easy to work with. The method returns the Future returned by recoverFn.

Without further ado, following are the Future methods that I promised. The name of each method is followed by its signature on the next line:

fallbackTo fallbackTo[U >: T](given: Future[U]): Future[U]

fallbackTo creates a new Future which inherits the value of the original Future if it completes successfully, or if the original Future fails, the result of the given Future if it completes successfully. If both the original and the given Futures fail, the resulting Future inherits the Throwable from the original Future.

Here is an example that shows this combinator in action, using dot notation and infix notation. The third usage of fallbackTo results in a failed Future because both the original Future and the fallback Future fail.

Scala code

object FutureFallbackTo extends App {
  import multi.futures.FutureFixtures._

  println("Dot notation:\n" + Await.result(badHostFuture.fallbackTo(defaultFuture), 30 minutes))
  println("\n\nInfix notation:\n" + Await.result(badHostFuture fallbackTo defaultFuture, 30 minutes))
  println("\n\n")
  Await.result(badHostFuture fallbackTo badPageFuture, 30 minutes)
}

Because Await.result is the last expression in the program, and the result is an Exception, the Exception is returned by the program.

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureFallbackTo"
Dot notation:
<!DOCTYPE html>
<html lang="en">
<head>
  <title>Online Scala Training</title>
  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
  <script src='https://cdn.jsdelivr.net/webjars/jquery/1.11.1/jquery.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-cookie/1.4.1/jquery.cookie.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-ui/1.11.1/jquery-ui.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/bootstrap/2.3.2/js/bootstrap....

Infix notation:
<!DOCTYPE html>
<html lang="en">
<head>
  <title>Online Scala Training</title>
  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
  <script src='https://cdn.jsdelivr.net/webjars/jquery/1.11.1/jquery.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-cookie/1.4.1/jquery.cookie.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-ui/1.11.1/jquery-ui.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/bootstrap/2.3.2/js/bootstrap....

Exception in thread "main" java.net.UnknownHostException: www.not-really-here.com ...

recover recover[U >: T](pf: PartialFunction[Throwable, U]) (implicit executor: ExecutionContext): Future[U]

recover creates a new Future from a partial function that receives any Throwable that the original failed Future might contain. If the original Future was successful, the new Future will have the same value as the original Future. If the partial function does not handle the original failed Future's Throwable, the new Future will inherit the unhandled Throwable from the original Future.

The example recovery strategies for this combinator all call doSomething, which is intended to suggest to you that you could write whatever code is appropriate in order to handle each type of Exception. Here is doSomething in all its glory; as you can see, it merely returns the String it receives.

Scala code

def doSomething(msg: String): String = msg

Here is an example of a Future that does not fail. Remember that the recovery strategy is passed into the show method, which is a higher-order function. In this example, the recover method is not invoked.

Scala code

show(goodUrlStr1, "no problem") {
  _.recover { case e: UnknownHostException => doSomething("Handled UnknownHostException") }
}

Output is:

Output

https://www.scalacourses.com; no problem, returning <!DOCTYPE html>
<html lang="en">
<head>
  <title>Online Scala Training</title>
  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
  <script src='https://cdn.jsdelivr.net/webjars/jquery/1.11.1/jquery.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-cookie/1.4.1/jquery.cookie.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-ui/1.11.1/jquery-ui.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/bootstrap/2.3.2/js/bootstrap....

Here is an example how recover handles a Future that fails:

Scala code

show(badHostUrlStr) {
  _.recover { case e: UnknownHostException => doSomething("Handled UnknownHostException") }
}

Output is:

Output

https://www.not-really-here.com; returning Handled UnknownHostException

Here is an example of using recover in a partial function that does not handle the Exception that is actually thrown.

Scala code

try {
  show(badHostUrlStr) {
    _.recover {
      case e: NoSuchElementException => throw new Exception("Should not need to handle NoSuchElementException")
    }
  }
} catch {
  case e: Exception =>
    println(s"Did not handle ${e.getClass.getName} exception for $badHostUrlStr")
}

Output is:

Output

Did not handle java.net.UnknownHostException exception for https://www.not-really-here.com

You can specify multiple recover clauses, each with a different partial function. The Scala compiler knows that MalformedURLException is never thrown so it issues an "unreachable code" warning for that case.

Scala code

show(badHostUrlStr, "handle 4 Exception types in 4 PartialFunctions using recover") {
  _.recover { case e: FileNotFoundException => doSomething("Handled FileNotFoundException") }
  .recover { case e: IOException           => doSomething("Handled IOException") }
  .recover { case e: MalformedURLException => doSomething("Handled MalformedURLException") }
  .recover { case e: UnknownHostException  => doSomething("Handled UnknownHostException") }
}

Output is:

Output

https://www.not-really-here.com; handle 4 Exception types in 4 PartialFunctions using recover, returning Handled IOException

Because each recover invocation above creates a new Future, there will be more context switching than if only one partial function with multiple case statements was supplied to one recover invocation, as shown in this next example. Again, the Scala compiler knows that MalformedURLException is never thrown so it issues an "unreachable code" warning for that case.

Scala code

show(badHostUrlStr, "handle 4 Exception types in one PartialFunction using recover") {
  _.recover {
    case e: FileNotFoundException => doSomething("Handled FileNotFoundException")
    case e: IOException           => doSomething("Handled IOException")
    case e: MalformedURLException => doSomething("Handled MalformedURLException")
    case e: UnknownHostException  => doSomething("Handled UnknownHostException")
  }
}

Output is:

Output

https://www.not-really-here.com; handle 4 Exception types in one PartialFunction using recover, returning Handled IOException

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureRecover"

Output is as shown above.

recoverWith recoverWith[U >: T](pf: PartialFunction[Throwable, Future[U]])(implicit executor: ExecutionContext): Future[U]

RecoverWith is like recover, but the partial function references another Future, instead of a value that will be wrapped up into a Future, and the value of that Future is used to create the returned Future if the recover clause is executed.

Here is an example of a successful Future, so the partial function supplied to the recoverWith combinator is not executed.

Scala code

show(goodUrlStr1, "no problem") {
  _.recoverWith { case e: UnknownHostException => Future.successful("Handled UnknownHostException") }
}

Output is:

Output

https://www.scalacourses.com; no problem, returning <!DOCTYPE html>
<html lang="en">
<head>
  <title>Online Scala Training</title>
  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
  <script src='https://cdn.jsdelivr.net/webjars/jquery/1.11.1/jquery.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-cookie/1.4.1/jquery.cookie.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-ui/1.11.1/jquery-ui.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/bootstrap/2.3.2/js/bootstrap....

Here is an example of a Future that fails, so the partial function supplied to the recoverWith combinator is invoked.

Scala code

show(badHostUrlStr) {
  _.recoverWith { case e: UnknownHostException => Future.successful("Handled UnknownHostException") }
}

Output is:

Scala code

https://www.not-really-here.com; returning Handled UnknownHostException
Did not handle java.net.UnknownHostException exception for https://www.not-really-here.com

Here is an example of using recoverWith with a partial function that does not handle the Exception that is actually thrown. In the following code, the Scala compiler knows that NoSuchElementException is never thrown so it issues an "unreachable code" warning for that case.

Scala code

try {
  show(badHostUrlStr) {
    _.recoverWith {
      case e: NoSuchElementException =>
        Future.failed(new Exception("Should not need to handle NoSuchElementException"))
    }
  }
} catch {
  case e: Exception =>
    println(s"Did not handle ${e.getClass.getName} exception for $badHostUrlStr")
}

Output is:

Scala code

Did not handle java.net.UnknownHostException exception for https://www.not-really-here.com

You can specify multiple recoverWith clauses, each with a different partial function. The Scala compiler knows that MalformedURLException is never thrown, so it issues an "unreachable code" warning for that case.

Scala code

show(badHostUrlStr, "handle 4 Exception types in 4 PartialFunctions using recoverWith") {
  _.recoverWith { case e: FileNotFoundException => defaultFuture}
  .recoverWith { case e: IOException           => defaultFuture}
  .recoverWith { case e: MalformedURLException => defaultFuture}
  .recoverWith { case e: UnknownHostException  => defaultFuture}
}

Output is:

Output

https://www.not-really-here.com; handle 4 Exception types in 1 PartialFunction using recoverWith, returning <!DOCTYPE html>
<html lang="en">
<head>
  <title>Online Scala Training</title>
  <meta http-equiv="Content-Type" content="text/html; charset=UTF-8" />
  <script src='https://cdn.jsdelivr.net/webjars/jquery/1.11.1/jquery.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-cookie/1.4.1/jquery.cookie.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/jquery-ui/1.11.1/jquery-ui.min.js'></script>
  <script src='https://cdn.jsdelivr.net/webjars/bootstrap/2.3.2/js/bootstrap....

The above code is equivalent to the following:

Scala code

show(badHostUrlStr, "handle 4 Exception types in 1 PartialFunction using recoverWith") {
  _.recoverWith {
    case e: FileNotFoundException => defaultFuture
    case e: IOException           => defaultFuture
    case e: MalformedURLException => defaultFuture
    case e: UnknownHostException  => defaultFuture
  }
}

Output is the same as before. You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureRecoverWith"

Output is as shown above.

Combinators for Individual Futures

collect collect[S](pf: PartialFunction[T, S]) (implicit executor: ExecutionContext): Future[S]

collect is like map, except it accepts a partial function, as we discussed in the Partial Functions lecture. If recover or recoverWith are chained after collect, values not handled by the partial function can be dealt with specially.

Scala code

val allUrlsWithFutures: List[(String, Future[String])] =
  urls(includeBad = true) map { url => url -> readUrlFuture(url) }

allUrlsWithFutures foreach {
  case tuple: (String, Future[String]) =>
    println(s"  Examining ${tuple._1}")
    tuple._2.collect {
      case content: String if content.toLowerCase.contains("scala") =>
        println(s"  Using Future.collect, scala was found in ${tuple._1}")
    }.recover { case _ =>
      println(s"  Future for ${tuple._1} failed")
    }
}

val futures = Future.sequence(allUrlsWithFutures.map { _._2 })
Await.ready(futures, 30 minutes)

The penultimate line creates a Future[List[String]] which is used by the last line to determine when to stop blocking the main thread so the program can exit. The Future.sequence combinator is described later in this lecture. You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureCollect"
Examining https://www.scalacourses.com
Examining https://www.scalacourses.com
Examining https://www.not-really-here.com
Examining https://scalacourses.com/noPageHere
Examining blah://scalacourses.com
Future for blah://scalacourses.com failed
Future for https://www.not_really_here.com failed
Future for https://scalacourses.com/noPageHere failed
Using Future.collect, scala was found in https://www.scalacourses.com

Note that the last line of the program is a call to Await.ready, which returns Unit. This becomes the returned value of the console application. If Await.result is called instead, the Exception from the first failed Future would always be reported.

filter filter(p: T ⇒ Boolean)(implicit executor: ExecutionContext): Future[T]

filter creates a new Future that contains the original Future's value if the predicate succeeds, or returns a new Future whose value is java.util.NoSuchElementException("Future.filter predicate is not satisfied") if the predicate fails. If recover or recoverWith are chained after filter, values that failed the predicate can be dealt with.

Scala code

1 to 10 foreach { i =>
  val future: Future[Int] = Future.successful(random.nextInt(100))
  val oddFuture: Future[Int]  = future filter { _ % 2 == 1 }
  val evenFuture: Future[Int] = future filter { _ % 2 == 0 }
  val oddOr24  = Await.result(oddFuture.recover      { case throwable: Throwable => 24},         30 seconds)
  val evenOr42 = Await.result(evenFuture.recover     { case throwable: Throwable => 42},         30 seconds)
  val all      = Await.result(evenFuture.recoverWith { case throwable: Throwable => oddFuture }, 30 seconds)
  println(f"$i%2.0f oddOr24=$oddOr24; evenOr42=$evenOr42; all=$all")
}

Either oddFuture or evenFuture must contain Failure(java.util.NoSuchElementException: Future.filter predicate is not satisfied). The oddOr24 and evenOr42 variables are created by calling recover on the potentially failed Futures, and provided success values 24 and 42, respectively, in place of failed Exceptions. Since we know that all numbers are either even or odd, we can define the all variable that will contain either the even or the odd number by calling evenFuture.recoverWith to provide the value of oddFuture in place of the NoSuchElementException.

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureFilter"
1: oddOr24=24; evenOr42=54; all=54
2: oddOr24=24; evenOr42=70; all=70
3: oddOr24=24; evenOr42=86; all=86
4: oddOr24=24; evenOr42=60; all=60
5: oddOr24=89; evenOr42=42; all=89
6: oddOr24=24; evenOr42=86; all=86
7: oddOr24=24; evenOr42=70; all=70
8: oddOr24=39; evenOr42=42; all=39
9: oddOr24=37; evenOr42=42; all=37
10: oddOr24=24; evenOr42=10; all=10

flatMap flatMap[S](f: T ⇒ Future[S])(implicit executor: ExecutionContext): Future[S]

flatMap is useful for applying an asynchronous method to a Future, and it yields a new Future containing the result. Asynchronous methods were introduced in the Futures & Promises lecture. Scala translates for-expressions into flatMap and map calls as discussed in the For-comprehensions With Futures lecture.

This code example requires a User case class, which has has two properties: name and privilege. We will use flatMap in a functional-style program that first pretends to retrieve a User instance from a database. Because this make-believe database has an asynchrous API, the User.findById method returns a Future[User]. User privilege is augmented by calling grantPrivilege, which is also an asynchronous call involving the make-believe database, so it also returns a Future[User]. Inside grantPrivilege, a new User instance is copied from the original instance and the extra privilege is added using the case class copy constructor. The copy constructor was discussed in the Case Classes lecture. There is a 30% chance that grantPrivilege will fail.

Scala code

case class User(name: String, privilege: List[String]) {
  /** simulate slow database access */
  def grantPrivilege(newPriv: String) = Future {
    Thread.sleep(random.nextInt(1000))
    if (System.currentTimeMillis % 3==0)
      throw new Exception("Unlucky time to grant privilege, not gonna do it!")
    copy(privilege = newPriv :: privilege)
  }
}

object User {
  private val userMap = Map(
    1L -> "Fred Flintstone",
    2L -> "Wilma Flintstone",
    3L -> "Barney Rubble",
    4L -> "Betty Rubble"
  )

  /** Simulate slow database access */
  def findById(id: Long): Future[User] = Future {
    Thread.sleep(random.nextInt(1000))
    User(userMap(id), Nil)
  }
}

Future.flatMap has the same semantics as the flatMap you already saw for Scala collections in the Combinators lecture. Inside the body of the flatMap, the contents of the monad are available and can be transformed. In this case we call User.grantPrivilege, pass a String describing the new privilege to be added to the User instance's privilege property, and obtain a Future[User]. If map had been used, a Future[Future[User]] would have been produced; instead, flatMap flattens the result to produce a Future[User], which is then passed along to the first andThen callback. andThen prints out the result, and the 2nd andThen callback signals that the program should terminate. This signalling pattern was introduced in the Future Callbacks lecture.

Scala code

val signal = Promise[String]()
val user: Future[User] = User.findById(random.nextInt(4)+1)
  .flatMap { _.grantPrivilege("student") }
  .andThen {
    case Success(value)     => println(s"""${value.name}'s privilege is now: ${value.privilege.mkString(", ")}.""")
    case Failure(throwable) => println(s"Problem augmenting student privilege: ${throwable.getMessage}")
  }.andThen { case _ => signal.success("All done") }
Await.ready(signal.future, 30 minutes)

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureFlatMap"

70% of the time output is something like:

Output

Fred Flintstone's privilege is now: student.

30% of the time output is:

Output

Problem augmenting student privilege: Unlucky time to grant privilege, not gonna do it!

foreach foreach[U](f: T ⇒ U) (implicit executor: ExecutionContext): Unit

Asynchronously processes the value in the Future once the value becomes available. This combinator is only useful for its side effects because it returns Unit. You might think that foreach is only useful when dealing with collections of futures, but instead foreach is a good way for an operation to be performed on a Future in a non-blocking, asynchronous manner.

Scala code

Future(factorial(12345)).foreach(println)

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureForeach"

Output is:

Scala code

34436424691867823916158... very long output ...

The code example in the courseNotes project uses a Promise to signal completion, otherwise no output would be seen.

map map[S](f: T ⇒ S) (implicit executor: ExecutionContext): Future[S]

map returns a new Future that contains the value of the original Future transformed according to the Function passed in. map is useful for applying a synchronous function in order to create a successful Future. map does not operate on a Future if it failed; instead, map passes along failed Futures unchanged. The following maps a Future[BigInt] to Future[Boolean]; the value of the new Future is true if factorial returned an even number.

Scala code

Future(factorial(12345)).map { _ % 2 == 0 }

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureMap"

The version of the program in the courseNotes project has been modified so it generates output before the main thread ends.

mapTo mapTo[S](implicit tag: ClassTag[S]): Future[S]

mapTo attempts to create a Future with a value type coerced to the given type S. If unsuccessful a ClassCastException is thrown. mapTo is useful for downcasting a Future's value, for example from Any to Int. mapTo is often used with Akka. This method should rarely be used since it defeats type safety.

This code example takes advantage of the fact that Future is covariant, so a Future[Int] can be cast to Future[Any]. mapTo does not change the Future's value, it just changes the declared type.

Scala code

val x: Future[Any] = Future.successful(1)
val y: Future[Int] = x.mapTo[Int]

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureMapTo"

transform transform[S](s: T ⇒ S, f: Throwable ⇒ Throwable) (implicit executor: ExecutionContext): Future[S]

transform is like map, except that in addition to the Function that transforms the value of a successful Future it also accepts a method that transforms the Throwable of a failed Future. This means that unlike map, which ignores failed Futures, transform operates on all Futures. If you don't intend to change the Throwable you should use map or collect instead of transform. You could think of transform with the identity transformation as 'transforming the Throwable'. We learned about the identity transformation in the Combinators lecture.

Here are two examples of how you can use transform on successful and failed Futures: the first example changes both the value of a successful Future and the Exception of a failed Future, while the second example preserves the success value by using the identity transform, and only changes the Exception of the failed Future.

Scala code

val f1 = Future.successful(random.nextInt(100))
  .filter(_%2==0)
  .transform(_/2, throwable => new Exception("Allergic to odd numbers...
achoo!", throwable))
  .andThen { case Success(value) => println(s"The (even) random number divided by 2 = $value") }

val f2 = Future(6/0)
  .transform(identity, throwable => new Exception("Something went wrong.", throwable))

val signal = Promise[String]()
Future.sequence(List(f1, f2))
  .andThen {
    case Success(value)     => println(value)
    case Failure(throwable) => println(throwable.getMessage)
  }.andThen { case _ => signal.success("All done") }
Await.ready(signal.future, 30 minutes)

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureTransform"

50% of the time output is:

Output

Allergic to odd numbers...
achoo!

The other 50% of the time output is something like:

Output

The (even) random number divided by 2 = 7
Something went wrong.

zip zip[U](that: Future[U]): Future[(T, U)]

zip is useful for associating a Future with an index or any arbitrary key. This combinator works with asychronous methods that return a Future. notice how similar this method is to flatMap; zip accepts a Future, whereas flatMap accepts a method that returns a Future.

This code example simulates a lottery, where users are retrieved from a (slow) database by specifying their id, and lucky numbers are (slowly) generated, using a database which ensures uniqueness. The zip method will create a Future[(User, Int)], where the Int is the lucky number.

Scala code

case class User(name: String, id: Long)

def getUser: Future[User] = Future { // simulate slow database access
  Thread.sleep(random.nextInt(1000))
  User("Fred Flintstone", 123)
}

def lotteryNumber: Future[Int] = Future { // simulate slow database access
  Thread.sleep(random.nextInt(1000))
  random.nextInt(1000000)
}

getUser zip lotteryNumber andThen {
  case Success(tuple) => println(s"User ${tuple._1.name} with id ${tuple._1.id} has lucky number ${tuple._2}.")
  case Failure(throwable) => println("Problem: " + throwable.getMessage)
} andThen { case _ => signal.success("All done") }

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureZip"

Typical output is:

Output

User Fred Flintstone with id 123 has lucky number 125876.

The version of the program in the courseNotes project has been modified so it generates output before the main thread ends.

Combinators for Collections of Futures

These methods are provided by the Future companion object. Some of them are similar to methods of the same name provided by the Iterable companion object, but they are non-blocking. The Scala compiler ensures that you won't mix up the methods provided by Future and Iterable, however.

find find[T](futures: Iterable[Future[T]]) (p: (T) => Boolean) (implicit executor: ExecutionContext): Future[Option[T]]

Completes with a result that satisfies the predicate; None is returned if no Future satisfies the predicate.

Normally, Scala Futures always run to completion; invoking the find combinator does not change that. If you need to optimize your program's use of Futures, see the Canceling Futures section in the Working With Collections of Futures lecture.

Notice that the andThen combinator has three cases: two for Success and one for None.

Scala code

val f1 = Future(factorial(12345))
val f2 = Future(factorial(23456))
val f3 = Future(factorial(34567))
val signal = Promise[String]()
Future
  .find(List(f1, f2, f3)) { _ % 2 == 0 }
  .andThen {
    case Success(Some(result)) => println(s"result = $result")
    case Success(None)         => println(s"No result was even")
    case Failure(throwable)    => println(throwable.getMessage)
  }.andThen { case _ => signal.success("All done") }
Await.ready(signal.future, 30 minutes)

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureFind"
result = 3443642469186... // very long number

firstCompletedOf firstCompletedOf[T](futures: IterableOnce[Future[T]]) (implicit executor: ExecutionContext): Future[T]

Section to complete. All Futures run to completion; Scala Future has no provision for terminating a Future once it starts, which is truly unfortunate. Twitter Futures (Finangle) do not have this limitation. The Working With Collections of Futures lecture shows a way of overcoming this limitation.

Scala code

val f1 = Future(factorial(12345))
val f2 = Future(factorial(23456))
val f3 = Future(factorial(34567))
Future.firstCompletedOf(List(f1, f2, f3)).andThen {
  case Success(result) => println(s"result = $result")
  case Failure(throwable) => println(throwable.getMessage)
}

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureFirstCompletedOf"

foldLeft foldLeft[T, R](futures: IterableOnce[Future[T]]) (zero: R) (op: (R, T) ⇒ R) (implicit executor: ExecutionContext): Future[R]

Prior to Scala 2.13 this method was called fold.

foldLeft iterates over the specified Futures and accumulates a result, using the given starting value zero. The computation of the result is performed on the thread used by the last Future to complete. The result will be the first failure of any of futures, or any failure encountered by op, or the result of the foldLeft.

Two examples of different foldLefts are shown below. The first foldLeft uses a lambda function (more precisely, a Function2[BigInt, BigInt, BigInt]) to create a new Future containing the summed value of all the Futures passed to it. The second foldLeft uses a named Function2[BigInt, BigInt, BigInt] called bigMax to create a new Future containing the largest value of all the Futures passed to it.

The program is written in such a way that the second foldLeft does not run until the first foldLeft has completed, however there is no reason the program could not be written such that both foldLefts occurred simultaneously since the computations are independent. The program shuts down after the second foldLeft.

Scala code

val f1 = Future(factorial(12345))
val f2 = Future(factorial(23456))
val f3 = Future(factorial(34567))
val futures = List(f1, f2, f3)

Future.foldLeft(futures)(BigInt(0))(_ + _) andThen {
    case Success(result)    => println(s"foldLeft addition result = $result")
    case Failure(throwable) => println(throwable.getMessage)
  } andThen {
    case _ =>
      val bigMax: (BigInt, BigInt) => BigInt =
        (x: BigInt, y: BigInt) => if (x>y) x else y
      Future
        .foldLeft(futures)(BigInt(0))(bigMax)
        .andThen {
          case Success(result)    => println(s"foldLeft max result = $result")
          case Failure(throwable) => println(throwable.getMessage)
        }
  }

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureFold"
foldLeft addition resu lt = 21364444 ... // very big number
foldLeft max result = 21364444 ... // very big number

reduceLeft reduceLeft[T, R >: T](futures: collection.immutable.Iterable[Future[T]]) (op: (R, T) => R) (implicit executor: ExecutionContext): Future[R]

reduceLeft is like foldLeft, except that the zero value is taken from the value of the first Future to complete, instead of being explicitly specified. Since this is normally what you want, reduceLeft is used more frequently than foldLeft.

This code example is just like the code example for foldLeft, with the foldLeft's zero parameter removed since it is not needed for reduceLeft.

Scala code

val f1 = Future(factorial(12345))
val f2 = Future(factorial(23456))
val f3 = Future(factorial(34567))
val futures = List(f1, f2, f3)
val bigMax = (x: BigInt, y: BigInt) => if (x>y) x else y
val signal = Promise[String]()

Future.reduceLeft(futures)(_+_)
  .andThen {
    case Success(result)    => println(s"Reduce addition result = $result")
    case Failure(throwable) => println(throwable.getMessage)
  } andThen {
    case _ =>
      Future.reduceLeft(futures)(bigMax)
        .andThen {
          case Success(result)    => println(s"Reduce max result = $result")
          case Failure(throwable) => println(throwable.getMessage)
        }.andThen { case _ => signal.success("All done") }
  }
Await.ready(signal.future, 30 minutes)

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureReduce"
reduce addition result = 2136444 ... // very big number
reduce max result = 21364444 ... // very big number

sequence sequence[A, CC[X] <: IterableOnce[X], To](in: CC[Future[A]]) (implicit bf: BuildFrom[CC[Future[A]], A, To], executor: ExecutionContext): Future[To]

sequence takes an IterableOnce[Future[A]] and returns an equivalent Future[IterableOnce[A]]. The new Future will complete when all Futures in the collection have completed. If any Future in the collection fails, that failure will be propagated to the new Future, however all Futures will run to completion even if one or more Futures should fail.

For example, sequence can be used to transform a List[Future[BigInt]] into a Future[List[BigInt]], so a callback can be triggered after all the original Futures have completed.

Scala code

val f1 = Future(factorial(12345))
val f2 = Future(factorial(23456))
val f3 = Future(factorial(34567))
val futures = List(f1, f2, f3)
val signal = Promise[String]()

Future.sequence(futures)
  .andThen {
    case Success(result)    => println(s"Sequence result = $result")
    case Failure(throwable) => println(throwable.getMessage)
  }.andThen { case _ => signal.success("All done") }
Await.ready(signal.future, 30 minutes)

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureSequence"
Reduce result = List(3443642469... very big number ...)

traverse traverse[A, B, M[X] <: IterableOnce[X]](in: M[A]) (fn: (A) => Future[B]) (implicit bf: BuildFrom[M[A], B, M[B]], executor: ExecutionContext): Future[M[B]]

traverse transforms an IterableOnce[A] into a Future[IterableOnce[B]] using the given Function1[A] => Future[B]. traverse is useful for performing a parallel and asynchronous map and is more flexible than sequence. The resulting Future only completes after all the Futures in the passed in collection complete. The order of the elements in the collection of the resulting Future's value will correspond to the order of the elements of the original collection. If any of the Futures fail, the resulting Future will contain the Throwable of the first failed Future, however all Futures will run to completion even if one or more fail.

For example, the following asynchronously applies factorial to all items of list in parallel, then prints a summary of all of the completed Futures.

Scala code

val signal = Promise[String]()
Future.traverse(List(12345, 234, 345)) { x =>
  Future(factorial(x)) andThen { case Success(value) => println(s"""factorial($x)=$value""") }
}.andThen {
  case Success(list)      => println(list.mkString("traverse results:\n  ", "\n  ", ""))
  case Failure(throwable) => println(throwable.getMessage)
}.andThen { case _ => signal.success("All done") }
Await.ready(signal.future, 30 minutes)

You can run this program by typing:

Shell

$ sbt "runMain multi.futures.FutureTraverse"

The output shows that each Future[BigInt] completes asynchronously, however the resulting Future[List[BigInt] only completes after every Future[BigInt] has completed.

Output

factorial(234)=22670150 ... // very big number
factorial(345)=24215638 ... // very big number
factorial(12345)=35135103 ... // very big number
traverse results:
  35135103 ...
// very big number
  22670150 ...
// very big number
  24215638 ...
// very big number

Previous lecture: For-comprehensions With Futures

Next lecture: Future Bad Habits and Exercise

© Copyright 1994-2024 Michael Slinn. All rights reserved.
If you would like to request to use this copyright-protected work in any manner, please send an email.

This website was made using Jekyll and Mike Slinn’s Jekyll Plugins.