package org.fungsi.concurrent;

import org.fungsi.Either;
import org.fungsi.Unit;
import org.fungsi.function.UnsafeFunction;

import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collector;

public final class Futures {
	private Futures() {}

	public static Future<Unit> unit() {
		return Future.constant(Unit.left());
	}

	public static <T> Future<T> success(T value) {
		return Future.constant(Either.success(value));
	}

	public static <T> Future<T> failure(Throwable cause) {
		return Future.constant(Either.failure(cause));
	}

	public static <T> Future<T> flatten(Either<Future<T>, Throwable> e) {
		return e.fold(Function.identity(), Futures::failure);
	}

	public static <T, R> Function<T, Future<R>> safe(UnsafeFunction<T, Future<R>> fn) {
		return fn.safeFunction().andThen(Futures::flatten);
	}

	public static <T> Future<List<T>> collect(List<Future<T>> futures) {
		Promise<List<T>> p = Promises.create();

		final AtomicInteger count = new AtomicInteger(futures.size());
		final List<T> values = new ArrayList<>();

		for (Future<T> fut : futures) {
			fut.onSuccess(value -> {
				values.add(value);

				if (count.decrementAndGet() == 0) {
					p.set(Either.success(values));
				}
			}).onFailure(cause -> {
				count.set(-1);
				p.set(Either.failure(cause));
			});
		}

		return p;
	}

	public static <T> Collector<Future<T>, ?, Future<List<T>>> collect() {
		return new Collector<Future<T>, List<Future<T>>, Future<List<T>>>() {
			@Override
			public Supplier<List<Future<T>>> supplier() {
				return ArrayList::new;
			}

			@Override
			public BiConsumer<List<Future<T>>, Future<T>> accumulator() {
				return List::add;
			}

			@Override
			public BinaryOperator<List<Future<T>>> combiner() {
				return (a, b) -> {
					a.addAll(b);
					return a;
				};
			}

			@Override
			public Function<List<Future<T>>, Future<List<T>>> finisher() {
				return Futures::collect;
			}

			@Override
			public Set<Characteristics> characteristics() {
				return EnumSet.of(Characteristics.UNORDERED);
			}
		};
	}
}

@@ -25,16 +25,64 @@
 rrent;%0A%0A
+import com.google.common.collect.ImmutableList;%0A
 import o
@@ -188,138 +188,9 @@
 til.
-ArrayList;%0Aimport java.util.EnumSet;%0Aimport java.util.List;%0Aimport java.util.Set;%0Aimport java.util.concurrent.atomic.AtomicInteger
+*
 ;%0Aim
@@ -987,24 +987,59 @@
 atten);%0A%09%7D%0A%0A
+    @SuppressWarnings(%22unchecked%22)%0A
 %09public stat
@@ -1096,18 +1096,234 @@
 ures) %7B%0A
-%09%09
+        if (futures.isEmpty()) %7B%0A            return Futures.success(ImmutableList.of());%0A        %7D%0A%0A        if (futures.size() == 1) %7B%0A            return futures.get(0).map(ImmutableList::of);%0A        %7D%0A%0A        final 
 Promise%3C
@@ -1332,16 +1332,22 @@
 st%3CT%3E%3E p
+romise
  = Promi
@@ -1365,56 +1365,96 @@
 );%0A%0A
-%09%09final AtomicInteger count = new AtomicInteger(
+        final Object%5B%5D lock = new Object%5B0%5D;%0A        final Object%5B%5D result = new Object%5B
 futu
@@ -1467,58 +1467,53 @@
 ze()
-);%0A%09%09final List%3CT%3E values = new ArrayList%3C%3E();%0A%0A%09%09
+%5D;%0A        final int%5B%5D index = %7B0%7D;%0A%0A        
 for 
@@ -1526,16 +1526,19 @@
 e%3CT%3E fut
+ure
  : futur
@@ -1547,14 +1547,77 @@
 ) %7B%0A
-%09%09%09fut
+            future%0A            .onFailure(promise::fail)%0A            
 .onS
@@ -1627,227 +1627,369 @@
 ess(
-value
+res
  -%3E %7B%0A
-%09%09%09%09values.add(value);%0A%0A%09%09%09%09if (count.decrementAndGet() == 0) %7B%0A%09%09%09%09%09p.set(Either.success(values));%0A%09%09%09%09%7D%0A%09%09%09%7D).onFailure(cause -%3E %7B%0A%09%09%09%09count.set(-1);%0A%09%09%09%09p.set(Either.failure(cause));%0A%09%09%09%7D);%0A%09%09%7D%0A%0A%09%09
+                if (promise.isDone()) return;%0A%0A                synchronized (lock) %7B%0A                    result%5Bindex%5B0%5D++%5D = res;%0A%0A                    if (index%5B0%5D %3E= result.length) %7B%0A                        promise.complete(Arrays.asList((T%5B%5D) result));%0A                    %7D%0A                %7D%0A            %7D)%0A            ;%0A        %7D%0A%0A        
 return p
 ;%0A%09%7D
@@ -1984,16 +1984,22 @@
 return p
+romise
 ;%0A%09%7D%0A%0A%09p

improve Futures#collect