package name.falgout.jeffrey.concurrent;

import static java.util.stream.Collectors.toList;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.function.Function;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import name.falgout.jeffrey.stream.future.FutureStream;
import name.falgout.jeffrey.stream.future.adapter.FutureStreamBridge;
import throwing.function.ThrowingSupplier;

public final class CompletableFutures {
  private CompletableFutures() {}

  /**
   * <p>
   * Creates a new {@code CompletableFuture} which will be completed by the given
   * {@code ThrowingSupplier}. If the supplier throws an exception, the {@code CompletableFuture}
   * will be {@link CompletableFuture#completeExceptionally(Throwable) completed exceptionally}.
   *
   * <p>
   * {@link ThrowingSupplier#get()} will be called from the
   * {@link CompletableFuture#runAsync(Runnable) default executor}.
   *
   * @param supplier
   *          a supplier returning the value for the {@code CompletableFuture}.
   * @return the new {@code CompletableFuture}
   */
  public static <T> CompletableFuture<T> newCompletableFuture(ThrowingSupplier<T, ?> supplier) {
    return newCompletableFuture(supplier,
        (Function<Runnable, CompletableFuture<Void>>) CompletableFuture::runAsync);
  }

  /**
   * <p>
   * Creates a new {@code CompletableFuture} which will be completed by the given
   * {@code ThrowingSupplier}. If the supplier throws an exception, the {@code CompletableFuture}
   * will be {@link CompletableFuture#completeExceptionally(Throwable) completed exceptionally}. *
   *
   * <p>
   * {@link ThrowingSupplier#get()} will be called from the
   * {@link CompletableFuture#runAsync(Runnable, Executor) given executor}.
   *
   * @param supplier
   *          a supplier returning the value for the {@code CompletableFuture}.
   * @param executor
   *          the executor to call {@link ThrowingSupplier#get() get} from.
   * @return the new {@code CompletableFuture}
   */
  public static <T> CompletableFuture<T> newCompletableFuture(ThrowingSupplier<T, ?> supplier,
      Executor executor) {
    return newCompletableFuture(supplier,
        (Function<Runnable, CompletableFuture<Void>>) r -> CompletableFuture.runAsync(r, executor));
  }

  private static <T> CompletableFuture<T> newCompletableFuture(ThrowingSupplier<T, ?> supplier,
      Function<Runnable, CompletableFuture<Void>> executor) {
    CompletableFuture<T> future = new CompletableFuture<>();
    executor.apply(() -> {
      try {
        future.complete(supplier.get());
      } catch (Throwable t) {
        future.completeExceptionally(t);
      }
    });
    return future;
  }

  /**
   * Creates a new immediately failed {@code CompletableFuture}.
   *
   * @param cause
   *          the exception to rethrow
   * @return a new {@link CompletableFuture#isCompletedExceptionally() exceptionally completed}
   *         {@code CompletableFuture}.
   */
  public static <T> CompletableFuture<T> failed(Throwable cause) {
    CompletableFuture<T> f = new CompletableFuture<>();
    f.completeExceptionally(cause);
    return f;
  }

  private static <T> CompletableFuture<T> newCompletableFuture(Future<T> f) {
    return f instanceof CompletableFuture ? (CompletableFuture<T>) f : newCompletableFuture(() -> {
      try {
        return f.get();
      } catch (ExecutionException e) {
        throw e.getCause();
      }
    });
  }

  @SafeVarargs
  public static <T> Future<List<T>> allOf(Future<T>... futures) {
    return allOf(Arrays.asList(futures));
  }

  public static <T> Future<List<T>> allOf(Iterable<? extends Future<T>> futures) {
    CompletableFuture<?>[] futuresArray = toArray(futures);
    CompletableFuture<Void> f = CompletableFuture.allOf(futuresArray);

    CompletableFuture<List<T>> all = f.thenApply(nil -> Arrays.stream(futuresArray)
        .map(CompletableFuture::join)
        .map(CompletableFutures.<T> unsafeCast())
        .collect(toList()));

    return all;
  }

  private static <T> CompletableFuture<?>[] toArray(Iterable<? extends Future<T>> futures) {
    return createStream(futures).map(CompletableFutures::newCompletableFuture).toArray(
        CompletableFuture<?>[]::new);
  }

  @SuppressWarnings("unchecked")
  private static <T> Function<Object, T> unsafeCast() {
    return o -> (T) o;
  }

  @SafeVarargs
  public static <T> Future<Optional<T>> anyOf(Future<T>... rest) {
    return anyOf(Arrays.asList(rest));
  }

  @SafeVarargs
  public static <T> Future<T> anyOf(Future<T> first, Future<T>... rest) {
    List<Future<T>> l = new ArrayList<>(1 + rest.length);
    l.add(first);
    l.addAll(Arrays.asList(rest));
    CompletableFuture<Optional<T>> f = newCompletableFuture(anyOf(l));

    return f.thenApply(Optional::get);
  }

  public static <T> Future<Optional<T>> anyOf(Iterable<? extends Future<T>> futures) {
    CompletableFuture<?>[] futuresArray = toArray(futures);
    if (futuresArray.length == 0) {
      return CompletableFuture.completedFuture(Optional.empty());
    }

    CompletableFuture<Object> f = CompletableFuture.anyOf(futuresArray);
    return f.thenApply(i -> Optional.of(i).map(unsafeCast()));
  }

  @SafeVarargs
  public static <T> FutureStream<T> stream(Future<T>... futures) {
    return stream(Arrays.asList(futures));
  }

  public static <T> FutureStream<T> stream(Iterable<? extends Future<T>> futures) {
    return FutureStreamBridge.of(createStream(futures));
  }

  private static <T> Stream<T> createStream(Iterable<T> elements) {
    return StreamSupport.stream(elements.spliterator(), false);
  }

  @SafeVarargs
  public static <T> FutureStream<T> stream(Executor executor, Future<T>... futures) {
    return stream(executor, Arrays.asList(futures));
  }

  public static <T> FutureStream<T> stream(Executor executor, Iterable<? extends Future<T>> futures) {
    return FutureStreamBridge.of(createStream(futures), executor);
  }
}

@@ -5008,32 +5008,631 @@
 nal::get);%0A  %7D%0A%0A
+  /**%0A   * Returns a new %7B@code Future%7D that is completed when any of the given %7B@code Future%7Ds complete,%0A   * with the same result. Otherwise, if it completed exceptionally, the returned %7B@code Future%7D%0A   * also does so, with a CompletionException holding this exception as its cause. If no%0A   * %7B@code Future%7Ds are provided, returns a %7B@code Future%7D which contains %7B@code Optional.empty()%7D.%0A   *%0A   * @param futures%0A   *          the futures%0A   * @return a new %7B@code Future%7D that is completed with the result or exception of any of the given%0A   *         %7B@code Future%7Ds when one completes%0A   */%0A
   public static 

Document the behavior of anyOf since it's different than CompletableFuture's