Futures (Scala)

Introduction

In Akka, a Future is a data structure used to retrieve the result of some concurrent operation. This operation is usually performed by an Actor or by the Dispatcher directly. This result can be accessed synchronously (blocking) or asynchronously (non-blocking).

Execution Contexts

In order to execute callbacks and operations, Futures need something called an ExecutionContext, which is very similar to a java.util.concurrent.Executor. if you have an ActorSystem in scope, it will use its default dispatcher as the ExecutionContext, or you can use the factory methods provided by the ExecutionContext companion object to wrap Executors and ExecutorServices, or even create your own.

import akka.dispatch.{ ExecutionContext, Promise }

implicit val ec = ExecutionContext.fromExecutorService(yourExecutorServiceGoesHere)

// Do stuff with your brand new shiny ExecutionContext
val f = Promise.successful("foo")

// Then shut your ExecutionContext down at some
// appropriate place in your program/application
ec.shutdown()

Use With Actors

There are generally two ways of getting a reply from an Actor: the first is by a sent message (actor ! msg), which only works if the original sender was an Actor) and the second is through a Future.

Using an Actor's ? method to send a message will return a Future. To wait for and retrieve the actual result the simplest method is:

import akka.dispatch.Await
import akka.pattern.ask
import akka.util.Timeout
import akka.util.duration._

implicit val timeout = Timeout(5 seconds)
val future = actor ? msg // enabled by the “ask” import
val result = Await.result(future, timeout.duration).asInstanceOf[String]

This will cause the current thread to block and wait for the Actor to 'complete' the Future with it's reply. Blocking is discouraged though as it will cause performance problems. The blocking operations are located in Await.result and Await.ready to make it easy to spot where blocking occurs. Alternatives to blocking are discussed further within this documentation. Also note that the Future returned by an Actor is a Future[Any] since an Actor is dynamic. That is why the asInstanceOf is used in the above sample. When using non-blocking it is better to use the mapTo method to safely try to cast a Future to an expected type:

import akka.dispatch.Future
import akka.pattern.ask

val future: Future[String] = ask(actor, msg).mapTo[String]

The mapTo method will return a new Future that contains the result if the cast was successful, or a ClassCastException if not. Handling Exceptions will be discussed further within this documentation.

Use Directly

A common use case within Akka is to have some computation performed concurrently without needing the extra utility of an Actor. If you find yourself creating a pool of Actors for the sole reason of performing a calculation in parallel, there is an easier (and faster) way:

import akka.dispatch.Await
import akka.dispatch.Future
import akka.util.duration._

val future = Future {
  "Hello" + "World"
}
val result = Await.result(future, 1 second)

In the above code the block passed to Future will be executed by the default Dispatcher, with the return value of the block used to complete the Future (in this case, the result would be the string: "HelloWorld"). Unlike a Future that is returned from an Actor, this Future is properly typed, and we also avoid the overhead of managing an Actor.

You can also create already completed Futures using the Promise companion, which can be either successes:

val future = Promise.successful("Yay!")

Or failures:

val otherFuture = Promise.failed[String](new IllegalArgumentException("Bang!"))

Functional Futures

Akka's Future has several monadic methods that are very similar to the ones used by Scala's collections. These allow you to create 'pipelines' or 'streams' that the result will travel through.

val f1 = Future {
  "Hello" + "World"
}
val f2 = f1 map { x ⇒
  x.length
}
val result = Await.result(f2, 1 second)
result must be(10)
f1.value must be(Some(Right("HelloWorld")))

val f1 = Future {
  "Hello" + "World"
}
val f2 = Promise.successful(3)
val f3 = f1 map { x ⇒
  f2 map { y ⇒
    x.length * y
  }
}

val f1 = Future {
  "Hello" + "World"
}
val f2 = Promise.successful(3)
val f3 = f1 flatMap { x ⇒
  f2 map { y ⇒
    x.length * y
  }
}
val result = Await.result(f3, 1 second)
result must be(30)

val future1 = Promise.successful(4)
val future2 = future1.filter(_ % 2 == 0)
val result = Await.result(future2, 1 second)
result must be(4)

val failedFilter = future1.filter(_ % 2 == 1).recover {
  case m: MatchError ⇒ 0 //When filter fails, it will have a MatchError
}
val result2 = Await.result(failedFilter, 1 second)
result2 must be(0) //Can only be 0 when there was a MatchError

val f = for {
  a ← Future(10 / 2) // 10 / 2 = 5
  b ← Future(a + 1) //  5 + 1 = 6
  c ← Future(a - 1) //  5 - 1 = 4
  if c > 3 // Future.filter
} yield b * c //  6 * 4 = 24

// Note that the execution of futures a, b, and c
// are not done in parallel.

val result = Await.result(f, 1 second)
result must be(24)

val f1 = ask(actor1, msg1)
val f2 = ask(actor2, msg2)

val a = Await.result(f1, 1 second).asInstanceOf[Int]
val b = Await.result(f2, 1 second).asInstanceOf[Int]

val f3 = ask(actor3, (a + b))

val result = Await.result(f3, 1 second).asInstanceOf[Int]

val f1 = ask(actor1, msg1)
val f2 = ask(actor2, msg2)

val f3 = for {
  a ← f1.mapTo[Int]
  b ← f2.mapTo[Int]
  c ← ask(actor3, (a + b)).mapTo[Int]
} yield c

val result = Await.result(f3, 1 second).asInstanceOf[Int]

// oddActor returns odd numbers sequentially from 1 as a List[Future[Int]]
val listOfFutures = List.fill(100)(akka.pattern.ask(oddActor, GetNext).mapTo[Int])

// now we have a Future[List[Int]]
val futureList = Future.sequence(listOfFutures)

// Find the sum of the odd numbers
val oddSum = Await.result(futureList.map(_.sum), 1 second).asInstanceOf[Int]
oddSum must be(10000)

val futureList = Future.traverse((1 to 100).toList)(x ⇒ Future(x * 2 - 1))
val oddSum = Await.result(futureList.map(_.sum), 1 second).asInstanceOf[Int]
oddSum must be(10000)

val futureList = Future.sequence((1 to 100).toList.map(x ⇒ Future(x * 2 - 1)))
val oddSum = Await.result(futureList.map(_.sum), 1 second).asInstanceOf[Int]
oddSum must be(10000)

val futures = for (i ← 1 to 1000) yield Future(i * 2) // Create a sequence of Futures
val futureSum = Future.fold(futures)(0)(_ + _)
Await.result(futureSum, 1 second) must be(1001000)

val futures = for (i ← 1 to 1000) yield Future(i * 2) // Create a sequence of Futures
val futureSum = Future.reduce(futures)(_ + _)
Await.result(futureSum, 1 second) must be(1001000)

Callbacks

Sometimes you just want to listen to a Future being completed, and react to that not by creating a new Future, but by side-effecting. For this Akka supports onComplete, onSuccess and onFailure, of which the latter two are specializations of the first.

future onSuccess {
  case "bar"     ⇒ println("Got my bar alright!")
  case x: String ⇒ println("Got some random string: " + x)
}

future onFailure {
  case ise: IllegalStateException if ise.getMessage == "OHNOES" ⇒
  //OHNOES! We are in deep trouble, do something!
  case e: Exception ⇒
  //Do something else
}

future onComplete {
  case Right(result) ⇒ doSomethingOnSuccess(result)
  case Left(failure) ⇒ doSomethingOnFailure(failure)
}

Define Ordering

Since callbacks are executed in any order and potentially in parallel, it can be tricky at the times when you need sequential ordering of operations. But there's a solution and it's name is andThen. It creates a new Future with the specified callback, a Future that will have the same result as the Future it's called on, which allows for ordering like in the following sample:

val result = Future { loadPage(url) } andThen {
  case Left(exception) ⇒ log(exception)
} andThen {
  case _ ⇒ watchSomeTV
}

Auxiliary Methods

Future fallbackTo combines 2 Futures into a new Future, and will hold the successful value of the second Future if the first Future fails.

val future4 = future1 fallbackTo future2 fallbackTo future3

You can also combine two Futures into a new Future that will hold a tuple of the two Futures successful results, using the zip operation.

val future3 = future1 zip future2 map { case (a, b) ⇒ a + " " + b }

Exceptions

Since the result of a Future is created concurrently to the rest of the program, exceptions must be handled differently. It doesn't matter if an Actor or the dispatcher is completing the Future, if an Exception is caught the Future will contain it instead of a valid result. If a Future does contain an Exception, calling Await.result will cause it to be thrown again so it can be handled properly.

It is also possible to handle an Exception by returning a different result. This is done with the recover method. For example:

val future = akka.pattern.ask(actor, msg1) recover {
  case e: ArithmeticException ⇒ 0
}

In this example, if the actor replied with a akka.actor.Status.Failure containing the ArithmeticException, our Future would have a result of 0. The recover method works very similarly to the standard try/catch blocks, so multiple Exceptions can be handled in this manner, and if an Exception is not handled this way it will behave as if we hadn't used the recover method.

You can also use the recoverWith method, which has the same relationship to recover as flatMap has to map, and is use like this:

val future = akka.pattern.ask(actor, msg1) recoverWith {
  case e: ArithmeticException        ⇒ Promise.successful(0)
  case foo: IllegalArgumentException ⇒ Promise.failed[Int](new IllegalStateException("All br0ken!"))
}