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;
                Promise<Optional<Cons<T>>> p = promises.get(index);
                p.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()) {
                                promises2.set(index, null);
                                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>(
                            removeNulls(promises2))));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        private static <T> List<T> removeNulls(List<T> list) {
            List<T> list2 = new ArrayList<T>();
            for (T t : list)
                if (t != null)
                    list2.add(t);
            return list2;
        }

        @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();
        }

    }
}

@@ -102,24 +102,97 @@
 .util.List;%0A
+import java.util.Set;%0Aimport java.util.concurrent.ConcurrentSkipListSet;%0A
 import java.
@@ -277,58 +277,8 @@
 an;%0A
-import java.util.concurrent.atomic.AtomicInteger;%0A
 impo
@@ -2153,59 +2153,86 @@
 nal 
-Atomic
+ConcurrentSkipListSet%3C
 Integer
+%3E
  co
-untTerminated = new Atomic
+mpleted = new ConcurrentSkipListSet%3C
 Integer
+%3E
 (
-0
 );%0A%0A
@@ -3068,33 +3068,27 @@
     
-promises2.set(index, null
+completed.add(index
 );%0A 
@@ -3128,37 +3128,20 @@
  (co
-untTerminated.incrementAndGet
+mpleted.size
 () =
@@ -3437,25 +3437,20 @@
  (co
-untTerminated.get
+mpleted.size
 () =
@@ -3534,16 +3534,18 @@
     else
+ %7B
 %0A       
@@ -3665,34 +3665,65 @@
     
-removeNulls(promises2))));
+copyExcluding(promises2, completed))));%0A                %7D
 %0A   
@@ -3877,32 +3877,63 @@
 %3CT%3E 
-removeNulls(List%3CT%3E list
+copyExcluding(List%3CT%3E list, Set%3CInteger%3E excludeIndexes
 ) %7B%0A
@@ -4001,49 +4001,90 @@
 or (
-T t : list)%0A                if (t != null
+int i = 0; i %3C list.size(); i++) %7B%0A                if (!excludeIndexes.contains(i)
 )%0A  
@@ -4111,19 +4111,43 @@
 st2.add(
-t);
+list.get(i));%0A            %7D
 %0A       

remove merged streams as they complete