Await extends EventStreamBase<Try<T>> implements AwaitingEventStream<Try<T>> { public static <T> AwaitingEventStream<Try<T>> awaitCompletionStage( EventStream<CompletionStage<T>> source, Executor clientThreadExecutor) { return new Await<>( source, (future, handler) -> addCompletionHandler(future, handler, clientThreadExecutor)); } private  Await(
            EventStream<F> source,
            BiConsumer<F, TriConsumer<T, Throwable, Boolean>> addCompletionHandler); static AwaitingEventStream<Try<T>> awaitCompletionStage(
            EventStream<CompletionStage<T>> source,
            Executor clientThreadExecutor); static AwaitingEventStream<Try<T>> awaitTask(
            EventStream<Task<T>> source); @Override final ObservableBooleanValue pendingProperty(); @Override final boolean isPending();  }
@Test public void testAwaitCompletionStage() throws InterruptedException, ExecutionException { CompletableFuture<List<Object>> emitted = new CompletableFuture<>(); EventSource<Integer> src = new EventSource<>(); AwaitingEventStream<Try<Integer>> squares = src.mapToCompletionStage(x -> async(x*x)).await(executor); executor.execute(() -> { List<Object> res = aggregate(EventStreams.merge( squares.map(Try::get), EventStreams.valuesOf(squares.pendingProperty()))); src.push(1); src.push(2); src.push(3); src.push(4); executor.execute(() -> executor.execute(() -> emitted.complete(res))); }); assertEquals(Arrays.asList(false, true, 1, 4, 9, 16, false), emitted.get()); }