package com.tradeshift.reaktive;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Various functions relating to CompletableFuture
 */
public class CompletableFutures {
    /**
     * Returns a future that has failed with the given exception.
     */
    public static <T> CompletableFuture<T> failedFuture(Throwable x) {
        CompletableFuture<T> promise = new CompletableFuture<T>();
        promise.completeExceptionally(x);
        return promise;
    }
    
    /**
     * Returns a future that completes when all the futures in the given collection have completed.
     */
    public static <T,U extends T> CompletableFuture<List<T>> sequence(Collection<CompletableFuture<U>> futures) {
        return sequence(futures.stream());
    }
    
    /**
     * Returns a future that completes when all the futures in the given stream have completed.
     */
    public static <T,U extends T> CompletableFuture<List<T>> sequence(Stream<CompletableFuture<U>> futures) {
        return CompletableFuture.supplyAsync(() -> futures.map(f -> f.join()).collect(Collectors.toList()));
    }
}

@@ -154,50 +154,75 @@
 eam.
-Collectors;%0Aimport java.util.stream.Stream
+Stream;%0A%0Aimport javaslang.Value;%0Aimport javaslang.collection.Vector
 ;%0A%0A/
@@ -527,17 +527,16 @@
 eFuture%3C
-T
 %3E();%0A   
@@ -1151,83 +1151,1061 @@
 ture
-.supplyAsync(() -%3E futures.map(f -%3E f.join()).collect(Collectors.toList()))
+s.%3CT,U%3Esequence(futures.collect(Vector.collector())).thenApply(v -%3E v.toJavaList());%0A    %7D%0A    %0A    /**%0A     * Returns a future that completes when all the futures in the given collection have completed.%0A     */%0A    public static %3CT,U extends T%3E CompletableFuture%3CVector%3CT%3E%3E sequence(Value%3CCompletableFuture%3CU%3E%3E futures) %7B%0A        CompletableFuture%3CVector%3CT%3E%3E result = new CompletableFuture%3C%3E();%0A%0A        // We don't join() the futures until allOf() has completed, since otherwise we'll be blocking%0A        // a thread on the common thread pool all the time we're waiting%0A        @SuppressWarnings(%7B %22unchecked%22, %22rawtypes%22 %7D)%0A        CompletableFuture%3C?%3E%5B%5D array = futures.toJavaArray((Class) CompletableFuture.class);%0A        CompletableFuture.allOf(array).whenComplete((done, exception) -%3E %7B%0A            if (exception != null) %7B%0A                result.completeExceptionally(exception);%0A            %7D else %7B%0A                result.complete(Vector.narrow(futures.map(f -%3E f.join()).toVector()));%0A            %7D%0A        %7D);%0A        %0A        return result
 ;%0A  

Allow to sequence futures javaslang collections, and fix blocking thread.