package com.tradeshift.reaktive;

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

import io.vavr.Value;
import io.vavr.collection.Vector;

/**
 * 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<>();
        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 CompletableFutures.<T,U>sequence(futures.collect(Vector.collector())).thenApply(v -> v.toJavaList());
    }
    
    /**
     * Returns a future that completes when all the futures in the given collection have completed.
     */
    public static <T,U extends T> CompletableFuture<Vector<T>> sequence(Value<CompletableFuture<U>> futures) {
        CompletableFuture<Vector<T>> result = new CompletableFuture<>();

        // We don't join() the futures until allOf() has completed, since otherwise we'll be blocking
        // a thread on the common thread pool all the time we're waiting
        @SuppressWarnings({ "unchecked", "rawtypes" })
        CompletableFuture<?>[] array = futures.toJavaArray((Class) CompletableFuture.class);
        CompletableFuture.allOf(array).whenComplete((done, exception) -> {
            if (exception != null) {
                result.completeExceptionally(exception);
            } else {
                result.complete(Vector.narrow(futures.map(f -> f.join()).toVector()));
            }
        });
        
        return result;
    }
}

@@ -126,16 +126,61 @@
 Future;%0A
+import java.util.concurrent.CompletionStage;%0A
 import j
@@ -203,16 +203,16 @@
 Stream;%0A
-
 %0Aimport 
@@ -847,32 +847,40 @@
 lection%3C
-CompletableFutur
+? extends CompletionStag
 e%3CU%3E%3E fu
@@ -1131,32 +1131,40 @@
 (Stream%3C
-CompletableFutur
+? extends CompletionStag
 e%3CU%3E%3E fu
@@ -1498,35 +1498,52 @@
 lue%3C
-CompletableFuture%3CU%3E%3E futur
+? extends CompletionStage%3CU%3E%3E completionStag
 es) 
@@ -1616,16 +1616,114 @@
 ure%3C%3E();
+%0A        Value%3CCompletableFuture%3CU%3E%3E futures = completionStages.map(f -%3E f.toCompletableFuture());
 %0A%0A      

Allow CompletionStage to be directly used in CompletableFutures