package fw.common.func;

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

/**
 * Abstract base class for implementations of {@link Flow}. Subclasses typically
 * override some methods for either efficiency, or for the concern they embrace.
 */
abstract class AbstractFlow<T> implements Flow<T> {
    /**
     * Method limited to just AbstractFlow and its subclasses. Forces a resolve
     * of the entire Flow, and results in a mutable list of the values in the
     * flow.
     */
    protected List<T> toMutableList() {
        return toMutableList(this);
    }

    protected static <T> List<T> toMutableList(Flow<T> flow) {
        List<T> result = new ArrayList<T>();

        for (T value : flow) {
            result.add(value);
        }

        return result;
    }

    public Iterator<T> iterator() {
        return new Iterator<T>() {
            private Flow<T> current = AbstractFlow.this;

            public boolean hasNext() {
                return !current.isEmpty();
            }

            public T next() {
                T next = current.first();

                current = current.rest();

                return next;
            }

            public void remove() {
                throw new UnsupportedOperationException("Flows are immutable.");
            }

        };
    }

    public Flow<T> concat(Collection<? extends T> collection) {
        return concat(F.flow(collection));
    }

    public <V extends T> Flow<T> append(V... values) {
        return concat(F.flow(values));
    }

    public Flow<T> concat(Flow<? extends T> other) {
        return F.lazy(new LazyConcat<T>(this, other));
    }

    /**
     * Subclasses may override this for efficiency.
     */
    public Flow<T> each(Consumer<? super T> worker) {
        assert worker != null;

        for (T value : this) {
            worker.accept(value);
        }

        return this;
    }

    public Flow<T> filter(Predicate<? super T> predicate) {
        assert predicate != null;

        return F.lazy(new LazyFilter<T>(predicate, this));
    }

    public <X> Flow<X> map(Function<T, X> mapper) {
        assert mapper != null;

        return F.lazy(new LazyMapper<T, X>(mapper, this));
    }

    public <X, Y> Flow<Y> map(BiFunction<T, X, Y> mapper, Flow<? extends X> flow) {
        assert mapper != null;
        assert flow != null;

        if (this.isEmpty() || flow.isEmpty())
            return F.emptyFlow();

        return F.lazy(new LazyMapper2<T, X, Y>(mapper, this, flow));
    }

    public <A> A reduce(Reducer<A, T> reducer, A initial) {
        assert reducer != null;

        A accumulator = initial;

        Flow<T> cursor = this;

        while (!cursor.isEmpty()) {
            accumulator = reducer.reduce(accumulator, cursor.first());
            cursor = cursor.rest();
        }

        return accumulator;
    }

    public <X> Flow<X> mapcat(Function<T, Flow<X>> mapper) {
        Flow<Flow<X>> flows = map(mapper);

        if (flows.isEmpty())
            return F.emptyFlow();

        return flows.rest().reduce(
                (accumulator, value) -> accumulator.concat(value),
                flows.first());
    }

    public Flow<T> remove(Predicate<? super T> predicate) {
        assert predicate != null;

        return filter(F.not(predicate));
    }

    public Flow<T> reverse() {
        if (isEmpty())
            return F.emptyFlow();

        return new ArrayFlow<T>(this).reverse();
    }

    public Flow<T> sort() {
        if (isEmpty())
            return F.emptyFlow();

        return new ArrayFlow<T>(this).sort();
    }

    public Flow<T> sort(Comparator<T> comparator) {
        if (isEmpty())
            return F.emptyFlow();

        return new ArrayFlow<T>(this).sort(comparator);
    }

    public List<T> toList() {
        if (isEmpty())
            return Collections.emptyList();

        return Collections.unmodifiableList(toMutableList());
    }

    @SuppressWarnings("unchecked")
    public T[] toArray(Class<T> type) {
        assert type != null;

        List<T> list = toMutableList();

        Object array = Array.newInstance(type, list.size());

        return list.toArray((T[]) array);
    }

    public int count() {
        if (isEmpty()) {
            return 0;
        }
        int count = 0;
        for (Flow<T> flow = this; flow != null && !flow.isEmpty(); flow = flow
                .rest()) {
            count++;
        }
        return count;
    }

    public Flow<T> take(int length) {
        return F.lazy(new LazyTake<T>(length, this));
    }

    public Flow<T> drop(int length) {
        assert length >= 0;

        if (length == 0)
            return this;

        return F.lazy(new LazyDrop<T>(length, this));
    }

    public Set<T> toSet() {
        Set<T> set = new HashSet<T>();

        each(F.addToCollection(set));

        return Collections.unmodifiableSet(set);
    }

    public <X> ZippedFlow<T, X> zipWith(Flow<X> otherFlow) {
        assert otherFlow != null;

        Flow<Tuple<T, X>> tupleFlow = F
                .lazy(new LazyZip<T, X>(this, otherFlow));

        return ZippedFlowImpl.create(tupleFlow);
    }

    public Flow<T> removeNulls() {
        return remove(F.isNull());
    }

    public boolean isEmpty() {
        // TODO Auto-generated method stub
        return false;
    }

    public T first() {
        // TODO Auto-generated method stub
        return null;
    }

    public Flow<T> rest() {
        // TODO Auto-generated method stub
        return null;
    }

    public Flow<T> interleave(Flow<T>... otherFlows) {
        List<Flow<T>> allFlows = new ArrayList<Flow<T>>(otherFlows.length + 1);
        allFlows.add(this);

        for (Flow<T> otherFlow : otherFlows) {
            allFlows.add(otherFlow);
        }

        return F.lazy(new Interleaver<T>(allFlows));
    }
}
