Test for combining different futures and handling exceptions along the way

recoverAndZipTest.java

public class FutureCombinationTest {

  private final ActorSystem actorSystem = ActorSystem.create(
  		"test",
			ConfigFactory.load().withValue("akka.actor.default-dispatcher.type",
					ConfigValueFactory.fromAnyRef("akka.testkit.CallingThreadDispatcherConfigurator")));

	private static class Result {

		private final boolean isComplete;
		private final List<Integer> numbers;

		public Result() {
			isComplete = false;
			numbers = Collections.emptyList();
		}

		public Result(final List<Integer> numbers) {
			this.numbers = numbers;
			isComplete = true;
		}

		public Result(final List<Integer> numbers, final boolean complete) {
			this.numbers = numbers;
			isComplete = complete;
		}

		public boolean isComplete() {
			return isComplete;
		}

		public List<Integer> getNumbers() {
			return numbers;
		}

		public Result combine(final Result other) {
			final boolean incomplete = !isComplete() || !other.isComplete();
			final ImmutableList<Integer> combinedList = new ImmutableList.Builder<Integer>().addAll(getNumbers()).addAll(other.getNumbers()).build();
			return new Result(combinedList, !incomplete);
		}

		@Override
		public String toString() {
			return isComplete + " " + numbers;
		}
	}

	@Test
	public void recoverAndZip() throws Exception {
		final MessageDispatcher dispatcher = actorSystem.dispatcher();
		final Future<List<Integer>> first = Futures.<List<Integer>> successful(Lists.newArrayList(1, 2, 3));
		final Future<List<Integer>> second = Futures.failed(new Exception("Foo"));
		final Future<List<Integer>> third = Futures.<List<Integer>> successful(Lists.newArrayList(4, 5, 6));

		final Future<Result> firstResult = mapAndRecover(first, dispatcher);
		final Future<Result> secondResult = mapAndRecover(second, dispatcher);
		final Future<Result> thirdResult = mapAndRecover(third, dispatcher);

		final Future<Result> finalResult = firstResult.zip(secondResult).map(new Mapper<Tuple2<Result, Result>, Result>() {
			@Override
			public Result apply(final Tuple2<Result, Result> parameter) {
				return parameter._1.combine(parameter._2);
			}

		}, dispatcher).zip(thirdResult).map(new Mapper<Tuple2<Result, Result>, Result>() {
			@Override
			public Result apply(final Tuple2<Result, Result> parameter) {
				return parameter._1.combine(parameter._2);
			}
		}, dispatcher);

		final Result result = Await.result(finalResult, Duration.create(1, TimeUnit.SECONDS));
		Assert.assertFalse(result.isComplete());
		Assert.assertEquals(Lists.newArrayList(1, 2, 3, 4, 5, 6), result.getNumbers());
	}

	private static Future<Result> mapAndRecover(final Future<List<Integer>> first, final ExecutionContext context) {
		return first.map(new Mapper<List<Integer>, Result>() {
			@Override
			public Result apply(final List<Integer> parameter) {
				return new Result(parameter);
			}
		}, context).recover(new Recover<Result>() {

			@Override
			public Result recover(final Throwable arg0) throws Throwable {
				return new Result();
			}
		}, context);
	}
}