Futures (Java)

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 ExecutionContexts class to wrap Executors and ExecutorServices, or even create your own.

import akka.dispatch.*;
import akka.util.Timeout;

import akka.dispatch.ExecutionContexts;
import akka.dispatch.ExecutionContextExecutorService;

      ExecutionContextExecutorService ec =
        ExecutionContexts.fromExecutorService(yourExecutorServiceGoesHere);

      //Use ec with your Futures
      Future<String> f1 = Futures.successful("foo", ec);

      // Then you shut the ec down somewhere at the end of your program/application.
      ec.shutdown();

Use with Actors

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

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

import akka.dispatch.*;
import akka.util.Timeout;

    Timeout timeout = new Timeout(Duration.parse("5 seconds"));
    Future<Object> future = Patterns.ask(actor, msg, timeout);
    String result = (String) Await.result(future, timeout.duration());

This will cause the current thread to block and wait for the UntypedActor to 'complete' the Future with it's reply. Blocking is discouraged though as it can cause performance problem. 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 UntypedActor is a Future<Object> since an UntypedActor is dynamic. That is why the cast to String is used in the above sample.

Use Directly

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

import akka.util.Duration;
import akka.japi.Function;
import java.util.concurrent.Callable;
import static akka.dispatch.Futures.future;
import static java.util.concurrent.TimeUnit.SECONDS;

    Future<String> f = future(new Callable<String>() {
      public String call() {
        return "Hello" + "World";
      }
    }, system.dispatcher());
    String result = (String) Await.result(f, Duration.create(1, SECONDS));

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 UntypedActor, this Future is properly typed, and we also avoid the overhead of managing an UntypedActor.

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

Future<String> future = Futures.successful("Yay!", system.dispatcher());

Or failures:

Future<String> otherFuture = Futures.failed(new IllegalArgumentException("Bang!"), system.dispatcher());

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.

import akka.util.Duration;
import akka.japi.Function;
import java.util.concurrent.Callable;
import static akka.dispatch.Futures.future;
import static java.util.concurrent.TimeUnit.SECONDS;

    Future<String> f1 = future(new Callable<String>() {
      public String call() {
        return "Hello" + "World";
      }
    }, system.dispatcher());

    Future<Integer> f2 = f1.map(new Mapper<String, Integer>() {
      public Integer apply(String s) {
        return s.length();
      }
    });

    int result = Await.result(f2, Duration.create(1, SECONDS));
    assertEquals(10, result);

Future<String> f1 = future(new Callable<String>() {
  public String call() throws Exception {
    Thread.sleep(100);
    return "Hello" + "World";
  }
}, system.dispatcher());

Future<Integer> f2 = f1.map(new Mapper<String, Integer>() {
  public Integer apply(String s) {
    return s.length();
  }
});

Future<String> f1 = future(new Callable<String>() {
  public String call() {
    return "Hello" + "World";
  }
}, system.dispatcher());

Thread.sleep(100);

Future<Integer> f2 = f1.map(new Mapper<String, Integer>() {
  public Integer apply(String s) {
    return s.length();
  }
});

Future<String> f1 = future(new Callable<String>() {
  public String call() {
    return "Hello" + "World";
  }
}, system.dispatcher());

Future<Integer> f2 = f1.flatMap(new Mapper<String, Future<Integer>>() {
  public Future<Integer> apply(final String s) {
    return future(new Callable<Integer>() {
      public Integer call() {
        return s.length();
      }
    }, system.dispatcher());
  }
});

Future<Integer> future1 = Futures.successful(4, system.dispatcher());
Future<Integer> successfulFilter = future1.filter(Filter.filterOf(new Function<Integer, Boolean>() {
  public Boolean apply(Integer i) {
    return i % 2 == 0;
  }
}));

Future<Integer> failedFilter = future1.filter(Filter.filterOf(new Function<Integer, Boolean>() {
  public Boolean apply(Integer i) {
    return i % 2 != 0;
  }
}));
//When filter fails, the returned Future will be failed with a scala.MatchError

import static akka.dispatch.Futures.sequence;

    //Some source generating a sequence of Future<Integer>:s
    Iterable<Future<Integer>> listOfFutureInts = source;

    // now we have a Future[Iterable[Integer]]
    Future<Iterable<Integer>> futureListOfInts = sequence(listOfFutureInts, system.dispatcher());

    // Find the sum of the odd numbers
    Future<Long> futureSum = futureListOfInts.map(new Mapper<Iterable<Integer>, Long>() {
      public Long apply(Iterable<Integer> ints) {
        long sum = 0;
        for (Integer i : ints)
          sum += i;
        return sum;
      }
    });

    long result = Await.result(futureSum, Duration.create(1, SECONDS));

import static akka.dispatch.Futures.traverse;

    //Just a sequence of Strings
    Iterable<String> listStrings = Arrays.asList("a", "b", "c");

    Future<Iterable<String>> futureResult = traverse(listStrings, new Function<String, Future<String>>() {
      public Future<String> apply(final String r) {
        return future(new Callable<String>() {
          public String call() {
            return r.toUpperCase();
          }
        }, system.dispatcher());
      }
    }, system.dispatcher());

    //Returns the sequence of strings as upper case
    Iterable<String> result = Await.result(futureResult, Duration.create(1, SECONDS));
    assertEquals(Arrays.asList("A", "B", "C"), result);

import akka.japi.Function2;
import static akka.dispatch.Futures.fold;


    //A sequence of Futures, in this case Strings
    Iterable<Future<String>> futures = source;

    //Start value is the empty string
    Future<String> resultFuture = fold("", futures, new Function2<String, String, String>() {
      public String apply(String r, String t) {
        return r + t; //Just concatenate
      }
    }, system.dispatcher());
    String result = Await.result(resultFuture, Duration.create(1, SECONDS));

import static akka.dispatch.Futures.reduce;


    //A sequence of Futures, in this case Strings
    Iterable<Future<String>> futures = source;

    Future<Object> resultFuture = reduce(futures, new Function2<Object, String, Object>() {
      public Object apply(Object r, String t) {
        return r + t; //Just concatenate
      }
    }, system.dispatcher());

    Object result = Await.result(resultFuture, Duration.create(1, SECONDS));

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(new OnSuccess<String>() {
  public void onSuccess(String result) {
    if ("bar" == result) {
      //Do something if it resulted in "bar"
    } else {
      //Do something if it was some other String
    }
  }
});

future.onFailure(new OnFailure() {
  public void onFailure(Throwable failure) {
    if (failure instanceof IllegalStateException) {
      //Do something if it was this particular failure
    } else {
      //Do something if it was some other failure
    }
  }
});

future.onComplete(new OnComplete<String>() {
    public void onComplete(Throwable failure, String result) {
        if (failure != null) {
            //We got a failure, handle it here
        } else {
            // We got a result, do something with it
        }
    }
});

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, and 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:

Future<String> future1 = Futures.successful("value", system.dispatcher()).andThen(new OnComplete<String>() {
    public void onComplete(Throwable failure, String result) {
        if (failure != null)
            sendToIssueTracker(failure);
    }
}).andThen(new OnComplete<String>() {
  public void onComplete(Throwable failure, String result) {
    if (result != null)
      sendToTheInternetz(result);
  }
});

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.

Future<String> future1 = Futures.failed(new IllegalStateException("OHNOES1"), system.dispatcher());
Future<String> future2 = Futures.failed(new IllegalStateException("OHNOES2"), system.dispatcher());
Future<String> future3 = Futures.successful("bar", system.dispatcher());
Future<String> future4 = future1.fallbackTo(future2).fallbackTo(future3); // Will have "bar" in this case
String result = Await.result(future4, Duration.create(1, SECONDS));
assertEquals("bar", result);

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.

Future<String> future1 = Futures.successful("foo", system.dispatcher());
Future<String> future2 = Futures.successful("bar", system.dispatcher());
Future<String> future3 = future1.zip(future2).map(new Mapper<scala.Tuple2<String, String>, String>() {
  public String apply(scala.Tuple2<String, String> zipped) {
    return zipped._1() + " " + zipped._2();
  }
});

String result = Await.result(future3, Duration.create(1, SECONDS));
assertEquals("foo bar", result);

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 UntypedActor 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:

Future<Integer> future = future(new Callable<Integer>() {
  public Integer call() {
    return 1 / 0;
  }
}, system.dispatcher()).recover(new Recover<Integer>() {
  public Integer recover(Throwable problem) throws Throwable {
    if (problem instanceof ArithmeticException)
      return 0;
    else
      throw problem;
  }
});
int result = Await.result(future, Duration.create(1, SECONDS));
assertEquals(result, 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:

Future<Integer> future = future(new Callable<Integer>() {
  public Integer call() {
    return 1 / 0;
  }
}, system.dispatcher()).recoverWith(new Recover<Future<Integer>>() {
  public Future<Integer> recover(Throwable problem) throws Throwable {
    if (problem instanceof ArithmeticException) {
      return future(new Callable<Integer>() {
        public Integer call() {
          return 0;
        }
      }, system.dispatcher());
    } else
      throw problem;
  }
});
int result = Await.result(future, Duration.create(1, SECONDS));
assertEquals(result, 0);