package pulley;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import pulley.transforms.Buffer;
import pulley.transforms.Concat;
import pulley.transforms.Filter;
import pulley.transforms.Reduce;
import pulley.transforms.ToList;
import pulley.util.Optional;

public class Stream<T> {

    private final Factory<Promise<Optional<Cons<T>>>> factory;

    public Stream(Factory<Promise<Optional<Cons<T>>>> factory) {
        this.factory = factory;
    }

    public Factory<? extends Promise<Optional<Cons<T>>>> factory() {
        return factory;
    }

    public static <T> Stream<T> stream(Factory<Promise<Optional<Cons<T>>>> factory) {
        return new Stream<T>(factory);
    }

    public <R> Stream<R> transform(final Transformer<T, R> transformer) {
        final StreamFactory<R> f = new StreamFactory<R>() {
            @Override
            public Promise<Optional<Cons<R>>> create() {
                final Promise<Optional<Cons<T>>> p = factory.create();
                return transformer.transform(p);
            }
        };
        return stream(f);
    }

    public Stream<T> scheduleOn(final Scheduler scheduler) {
        return transform(new Transformer<T, T>() {

            @Override
            public Promise<Optional<Cons<T>>> transform(Promise<Optional<Cons<T>>> promise) {
                return new ScheduledPromise<Optional<Cons<T>>>(promise, scheduler);
            }
        });
    }

    public <R> Stream<R> map(final F1<? super T, ? extends R> f) {
        final F1<Optional<Cons<T>>, Optional<Cons<R>>> g = F.optional(F.cons(f));
        Transformer<T, R> transformer = new Transformer<T, R>() {
            @Override
            public StreamPromise<R> transform(final Promise<Optional<Cons<T>>> p) {
                return new StreamPromise<R>() {
                    @Override
                    public Optional<Cons<R>> get() {
                        return g.call(p.get());
                    }

                    @Override
                    public A0 closeAction() {
                        return p.closeAction();
                    }

                    @Override
                    public Scheduler scheduler() {
                        return p.scheduler();
                    }
                };
            }
        };
        return transform(transformer);
    }

    public Stream<T> doOnTerminate(A0 action) {
        // TODO
        return Util.notImplemented();
    }

    public Stream<List<T>> toList() {
        return ToList.toList(this);
    }

    public Stream<List<T>> buffer(final int size) {
        return Buffer.buffer(this, size);
    }

    public <R> Stream<R> reduce(R initial, F2<? super R, ? super T, ? extends R> reducer) {
        return Reduce.reduce(this, initial, reducer);
    }

    public Stream<Integer> count() {
        return reduce(0, new F2<Integer, T, Integer>() {

            @Override
            public Integer call(Integer count, T t) {
                return count + 1;
            }
        });
    }

    public Stream<T> filter(final F1<? super T, Boolean> predicate) {
        return Filter.filter(this, predicate);
    }

    public Stream<T> concatWith(final Stream<T> stream) {
        return Concat.concat(this, stream);
    }

    public <R> Stream<R> flatMap(F1<T, Stream<R>> f) {
        return Streams.merge(map(f));
    }

    public void forEach(A1<? super T> action) {
        forEach(factory.create(), action);
    }

    public static <T> void forEach(Promise<Optional<Cons<T>>> promise, final A1<? super T> action) {
        Optional<Promise<Optional<Cons<T>>>> p = Optional.of(promise);
        do {
            p = Promises.performActionAndAwaitCompletion(p.get(), action);
        } while (p.isPresent());
    }

    public void forEach() {
        forEach(Actions.doNothing1());
    }

    public T single() {
        final Promise<Optional<Cons<T>>> p = factory.create();
        final List<T> list = Collections.synchronizedList(new ArrayList<T>());
        A1<T> addToList = Actions.addToList(list);
        final Optional<Promise<Optional<Cons<T>>>> p2 = Promises.performActionAndAwaitCompletion(p,
                addToList);
        if (list.size() == 0) {
            throw new RuntimeException("expected one item but no items emitted");
        } else {
            Promises.performActionAndAwaitCompletion(p2.get(), addToList);
            if (list.size() > 1)
                throw new RuntimeException("expected one item but more than one emitted");
            else
                return list.get(0);
        }
    }
}
@@ -187,24 +187,54 @@
 rms.Filter;%0A
+import pulley.transforms.Map;%0A
 import pulle
@@ -1559,832 +1559,30 @@
     
-final F1%3COptional%3CCons%3CT%3E%3E, Optional%3CCons%3CR%3E%3E%3E g = F.optional(F.cons(f));%0A        Transformer%3CT, R%3E transformer = new Transformer%3CT, R%3E() %7B%0A            @Override%0A            public StreamPromise%3CR%3E transform(final Promise%3COptional%3CCons%3CT%3E%3E%3E p) %7B%0A                return new StreamPromise%3CR%3E() %7B%0A                    @Override%0A                    public Optional%3CCons%3CR%3E%3E get() %7B%0A                        return g.call(p.get());%0A                    %7D%0A%0A                    @Override%0A                    public A0 closeAction() %7B%0A                        return p.closeAction();%0A                    %7D%0A%0A                    @Override%0A                    public Scheduler scheduler() %7B%0A                        return p.scheduler();%0A                    %7D%0A                %7D;%0A            %7D%0A        %7D;%0A        return transform(transformer
+return Map.map(this, f
 );%0A 

move Map