package net.ranides.assira.collection.query.derived;

import net.ranides.assira.collection.iterators.IteratorUtils;
import net.ranides.assira.collection.query.CQuery;
import net.ranides.assira.collection.query.base.CQAbstractFilter;
import net.ranides.assira.collection.query.support.BaseState;
import net.ranides.assira.functional.Predicates;

import java.util.Iterator;
import java.util.function.Predicate;
import java.util.stream.Stream;

/**
 * Adapter query: it filters provided stream using predicate.
 * Use by {@link CQuery#filterEach(Predicates.EachPredicate)} operation.
 *
 * Supports fast iterator and stream if source supports it.
 * Does not support other fast operations.
 *
 * It supports stateless predicates only
 *
 * @param <T> T
 */
public class CQFilterEach<T> extends CQAbstractFilter<T, T> {

    private final Predicates.EachPredicate<? super T> p;

    /**
     * Creates new filtered query.
     * It assumes that predicate is stateless, and is reused by all terminal operations.
     *
     * @param source source
     * @param predicate predicate
     */
    public CQFilterEach(CQuery<T> source, Predicates.EachPredicate<? super T> predicate) {
        super(source);
        this.p = predicate;
    }

    @Override
    public CQuery<T> parallel() {
        return isParallel() ? this : new CQFilterEach<>(source.parallel(), p);
    }

    @Override
    public CQuery<T> sequential() {
        return isParallel() ? new CQFilterEach<>(source.sequential(), p) : this;
    }

    @Override
    public Stream<T> stream() {
        BaseState.Index index = BaseState.newIndex(source);
        return source.stream().filter(v -> p.test(index.next(), v));
    }

    @Override
    public Iterator<T> iterator() {
        return IteratorUtils.filterEach(source.iterator(), p);
    }

    @Override
    public boolean whileEach(Predicate<? super T> consumer) {
        return source.whileEach((i, v) -> !p.test(i, v) || consumer.test(v));
    }

}
