package pulley.streams;

import static pulley.Stream.stream;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import pulley.A0;
import pulley.Cons;
import pulley.Factory;
import pulley.Promise;
import pulley.Promises;
import pulley.Scheduler;
import pulley.Schedulers;
import pulley.Stream;
import pulley.Util;
import pulley.util.Optional;

public class Merge {

    public static <T> Stream<T> create(final List<Stream<T>> list) {
        return stream(factory(list));
    }

    public static <T> Stream<T> create(final Stream<Stream<T>> streams) {
        return Util.notImplemented();
    }

    private static <T> Factory<Promise<Optional<Cons<T>>>> factory(final List<Stream<T>> streams) {
        return new Factory<Promise<Optional<Cons<T>>>>() {
            @Override
            public Promise<Optional<Cons<T>>> create() {
                List<Promise<Optional<Cons<T>>>> promises = new ArrayList<Promise<Optional<Cons<T>>>>(
                        streams.size());
                for (Stream<T> stream : streams) {
                    promises.add(stream.factory().create());
                }
                return new MergePromise<T>(promises);
            }
        };
    }

    private static class MergePromise<T> implements Promise<Optional<Cons<T>>> {

        private final List<Promise<Optional<Cons<T>>>> promises;

        MergePromise(List<Promise<Optional<Cons<T>>>> promises) {
            this.promises = promises;
        }

        @Override
        public Optional<Cons<T>> get() {
            // blocking !!!!
            final List<Promise<Optional<Cons<T>>>> promises2 = new ArrayList<Promise<Optional<Cons<T>>>>(
                    promises);
            final AtomicBoolean found = new AtomicBoolean(false);
            final AtomicReference<Optional<Cons<T>>> value = new AtomicReference<Optional<Cons<T>>>();
            final CountDownLatch latch = new CountDownLatch(1);
            final AtomicInteger countTerminated = new AtomicInteger(0);

            for (int i = 0; i < promises.size(); i++) {
                final int index = i;
                promises.get(index).scheduler().schedule(new A0() {
                    @Override
                    public void call() {
                        if (!found.get()) {
                            promises2.set(index, Promises.cache(promises2.get(index)));
                            Optional<Cons<T>> t = promises2.get(index).get();
                            if (t.isPresent() && found.compareAndSet(false, true)) {
                                value.set(t);
                                promises2.set(index, t.get().tail());
                                latch.countDown();
                            } else if (!t.isPresent()) {
                                if (countTerminated.incrementAndGet() == promises.size()) {
                                    latch.countDown();
                                }
                            }
                        }
                    }
                });
            }
            try {
                latch.await();
                if (countTerminated.get() == promises.size())
                    return Optional.absent();
                else
                    return Optional.of(Cons.cons(value.get().get().head(), new MergePromise<T>(
                            promises2)));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public A0 closeAction() {
            return new A0() {
                @Override
                public void call() {
                    for (Promise<Optional<Cons<T>>> promise : promises)
                        promise.closeAction().call();
                }
            };
        }

        @Override
        public Scheduler scheduler() {
            return Schedulers.immediate();
        }

    }
}

@@ -2294,27 +2294,77 @@
     
-promises.get(index)
+Promise%3COptional%3CCons%3CT%3E%3E%3E p = promises.get(index);%0A                p
 .sch
@@ -2978,24 +2978,84 @@
 resent()) %7B%0A
+                                promises2.set(index, null);%0A
             
@@ -3629,32 +3629,44 @@
                 
+removeNulls(
 promises2)));%0A  
@@ -3661,16 +3661,17 @@
 ises2)))
+)
 ;%0A      
@@ -3754,24 +3754,24 @@
 ception(e);%0A
-
             
@@ -3775,32 +3775,274 @@
    %7D%0A        %7D%0A%0A
+        private static %3CT%3E List%3CT%3E removeNulls(List%3CT%3E list) %7B%0A            List%3CT%3E list2 = new ArrayList%3CT%3E();%0A            for (T t : list)%0A                if (t != null)%0A                    list2.add(t);%0A            return list2;%0A        %7D%0A%0A
         @Overrid

remove merged streams as they complete