SafeQueueingFuturePool { public synchronized Future<T> apply(final Function0<T> fn) { Preconditions.checkNotNull(fn); if (closed) { return Future.exception(new RejectedExecutionException("Operation submitted to closed SafeQueueingFuturePool")); } ++outstanding; queue.add(fn); Future<T> result = orderedFuturePool.apply(new Function0<T>() { @Override public T apply() { return queue.poll().apply(); } @Override public String toString() { return fn.toString(); } }).ensure(new Function0<BoxedUnit>() { public BoxedUnit apply() { if (decrOutstandingAndCheckDone()) { applyAll(); } return null; } }); return result; } SafeQueueingFuturePool(FuturePool orderedFuturePool); synchronized Future<T> apply(final Function0<T> fn); void close(); synchronized int size();  }
@Test public void testSimpleSuccess() throws Exception { TestFuturePool<Void> pool = new TestFuturePool<Void>(); final AtomicBoolean result = new AtomicBoolean(false); Future<Void> future = pool.wrapper.apply(new Function0<Void>() { public Void apply() { result.set(true); return null; } }); Await.result(future); assertTrue(result.get()); pool.shutdown(); }
@Test public void testSimpleFailure() throws Exception { TestFuturePool<Void> pool = new TestFuturePool<Void>(); Future<Void> future = pool.wrapper.apply(new Function0<Void>() { public Void apply() { throw new RuntimeException("failed"); } }); try { Await.result(future); fail("should have thrown"); } catch (Exception ex) { } pool.shutdown(); }