package stream;

import function.*;
import java.util.HashSet;
import java.util.Set;

/**
 * @Author xiongyx
 * @Date 2019/3/6
 *
 * stream实现
 */
public class MyStream<T> implements Stream<T> {

    /**
     * 流的头部
     * */
    private T head;

    /**
     * 流的下一项求值函数
     * */
    private NextItemEvalProcess nextItemEvalProcess;

    /**
     * 是否是流的结尾
     * */
    private boolean isEnd;

    public static class Builder<T>{
        private MyStream<T> target;

        public Builder() {
            this.target = new MyStream<>();
        }

        public Builder<T> head(T head){
            target.head = head;
            return this;
        }

        Builder<T> isEnd(boolean isEnd){
            target.isEnd = isEnd;
            return this;
        }

        public Builder<T> nextItemEvalProcess(NextItemEvalProcess nextItemEvalProcess){
            target.nextItemEvalProcess = nextItemEvalProcess;
            return this;
        }

        public MyStream<T> build(){
            return target;
        }
    }

    //=================================API接口实现==============================

    @Override
    public <R> MyStream<R> map(Function<R, T> mapper) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
                ()->{
                    MyStream myStream = lastNextItemEvalProcess.eval();
                    return map(mapper, myStream);
                }
        );

        // 求值链条 加入一个新的process map
        return new MyStream.Builder<R>()
                .nextItemEvalProcess(this.nextItemEvalProcess)
                .build();
    }

    @Override
    public <R> MyStream<R> flatMap(Function<? extends MyStream<R>,T> mapper) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
            ()->{
                MyStream myStream = lastNextItemEvalProcess.eval();
                return flatMap(mapper, Stream.makeEmptyStream(), myStream);
            }
        );

        // 求值链条 加入一个新的process map
        return new MyStream.Builder<R>()
            .nextItemEvalProcess(this.nextItemEvalProcess)
            .build();
    }

    @Override
    public MyStream<T> filter(Predicate<T> predicate) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
                ()-> {
                    MyStream myStream = lastNextItemEvalProcess.eval();
                    return filter(predicate, myStream);
                }
        );

        // 求值链条 加入一个新的process filter
        return this;
    }

    @Override
    public MyStream<T> limit(int n) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
                ()-> {
                    MyStream myStream = lastNextItemEvalProcess.eval();
                    return limit(n, myStream);
                }
        );

        // 求值链条 加入一个新的process limit
        return this;
    }

    @Override
    public MyStream<T> distinct() {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
            ()-> {
                MyStream myStream = lastNextItemEvalProcess.eval();
                return distinct(new HashSet<>(), myStream);
            }
        );

        // 求值链条 加入一个新的process limit
        return this;
    }

    @Override
    public MyStream<T> peek(ForEach<T> consumer) {
        NextItemEvalProcess lastNextItemEvalProcess = this.nextItemEvalProcess;
        this.nextItemEvalProcess = new NextItemEvalProcess(
            ()-> {
                MyStream myStream = lastNextItemEvalProcess.eval();
                return peek(consumer,myStream);
            }
        );

        // 求值链条 加入一个新的process peek
        return this;
    }

    @Override
    public void forEach(ForEach<T> consumer) {
        // 终结操作 直接开始求值
        forEach(consumer,this.eval());
    }

    @Override
    public <R> R reduce(R initVal, BiFunction<R, R, T> accumulator) {
        // 终结操作 直接开始求值
        return reduce(initVal,accumulator,this.eval());
    }

    @Override
    public <R, A> R collect(Collector<T, A, R> collector) {
        // 终结操作 直接开始求值
        A result = collect(collector,this.eval());

        // 通过finish方法进行收尾
        return collector.finisher().apply(result);
    }

    @Override
    public T max(Comparator<T> comparator) {
        // 终结操作 直接开始求值
        MyStream<T> eval = this.eval();

        if(eval.isEmptyStream()){
            return null;
        }else{
            return max(comparator,eval,eval.head);
        }
    }

    @Override
    public T min(Comparator<T> comparator) {
        // 终结操作 直接开始求值
        MyStream<T> eval = this.eval();

        if(eval.isEmptyStream()){
            return null;
        }else{
            return min(comparator,eval,eval.head);
        }
    }

    @Override
    public int count() {
        // 终结操作 直接开始求值
        return count(this.eval(),0);
    }

    @Override
    public boolean anyMatch(Predicate<? super T> predicate) {
        // 终结操作 直接开始求值
        return anyMatch(predicate,this.eval());
    }

    @Override
    public boolean allMatch(Predicate<? super T> predicate) {
        // 终结操作 直接开始求值
        return allMatch(predicate,this.eval());
    }

    //===============================私有方法====================================

    /**
     * 递归函数 配合API.map
     * */
    private static <R,T> MyStream<R> map(Function<R, T> mapper, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        R head = mapper.apply(myStream.head);

        return new MyStream.Builder<R>()
                .head(head)
                .nextItemEvalProcess(new NextItemEvalProcess(()->map(mapper, myStream.eval())))
                .build();
    }

    /**
     * 递归函数 配合API.flatMap
     * */
    private static <R,T> MyStream<R> flatMap(Function<? extends MyStream<R>,T> mapper, MyStream<R> headMyStream, MyStream<T> myStream){
        if(headMyStream.isEmptyStream()){
            if(myStream.isEmptyStream()){
                return Stream.makeEmptyStream();
            }else{
                T outerHead = myStream.head;
                MyStream<R> newHeadMyStream = mapper.apply(outerHead);

                return flatMap(mapper, newHeadMyStream.eval(), myStream.eval());
            }
        }else{
            return new MyStream.Builder<R>()
                        .head(headMyStream.head)
                        .nextItemEvalProcess(new NextItemEvalProcess(()-> flatMap(mapper, headMyStream.eval(), myStream)))
                        .build();
        }
    }

    /**
     * 递归函数 配合API.filter
     * */
    private static <T> MyStream<T> filter(Predicate<T> predicate, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        if(predicate.satisfy(myStream.head)){
            return new Builder<T>()
                    .head(myStream.head)
                    .nextItemEvalProcess(new NextItemEvalProcess(()->filter(predicate, myStream.eval())))
                    .build();
        }else{
            return filter(predicate, myStream.eval());
        }
    }

    /**
     * 递归函数 配合API.limit
     * */
    private static <T> MyStream<T> limit(int num, MyStream<T> myStream){
        if(num == 0 || myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        return new MyStream.Builder<T>()
                .head(myStream.head)
                .nextItemEvalProcess(new NextItemEvalProcess(()->limit(num-1, myStream.eval())))
                .build();
    }

    /**
     * 递归函数 配合API.distinct
     * */
    private static <T> MyStream<T> distinct(Set<T> distinctSet,MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        if(!distinctSet.contains(myStream.head)){
            // 加入集合
            distinctSet.add(myStream.head);

            return new Builder<T>()
                .head(myStream.head)
                .nextItemEvalProcess(new NextItemEvalProcess(()->distinct(distinctSet, myStream.eval())))
                .build();
        }else{
            return distinct(distinctSet, myStream.eval());
        }
    }

    /**
     * 递归函数 配合API.peek
     * */
    private static <T> MyStream<T> peek(ForEach<T> consumer,MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return Stream.makeEmptyStream();
        }

        consumer.apply(myStream.head);

        return new MyStream.Builder<T>()
            .head(myStream.head)
            .nextItemEvalProcess(new NextItemEvalProcess(()->peek(consumer, myStream.eval())))
            .build();
    }

    /**
     * 递归函数 配合API.forEach
     * */
    private static <T> void forEach(ForEach<T> consumer, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return;
        }

        consumer.apply(myStream.head);
        forEach(consumer, myStream.eval());
    }

    /**
     * 递归函数 配合API.reduce
     * */
    private static <R,T> R reduce(R initVal, BiFunction<R,R,T> accumulator, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return initVal;
        }

        T head = myStream.head;
        R result = reduce(initVal,accumulator, myStream.eval());

        return accumulator.apply(result,head);
    }

    /**
     * 递归函数 配合API.collect
     * */
    private static <R, A, T> A collect(Collector<T, A, R> collector, MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            return collector.supplier().get();
        }

        T head = myStream.head;
        A tail = collect(collector, myStream.eval());

        return collector.accumulator().apply(tail,head);
    }

    /**
     * 递归函数 配合API.max
     * */
    private static <T> T max(Comparator<T> comparator, MyStream<T> myStream, T max){
        if(myStream.isEnd){
            return max;
        }

        T head = myStream.head;
        // head 和 max 进行比较
        if(comparator.compare(head,max) > 0){
            // head 较大 作为新的max传入
            return max(comparator, myStream.eval(),head);
        }else{
            // max 较大 不变
            return max(comparator, myStream.eval(),max);
        }
    }

    /**
     * 递归函数 配合API.min
     * */
    private static <T> T min(Comparator<T> comparator, MyStream<T> myStream, T min){
        if(myStream.isEnd){
            return min;
        }

        T head = myStream.head;
        // head 和 min 进行比较
        if(comparator.compare(head,min) < 0){
            // head 较小 作为新的min传入
            return min(comparator, myStream.eval(),head);
        }else{
            // min 较小 不变
            return min(comparator, myStream.eval(),min);
        }
    }

    /**
     * 递归函数 配合API.count
     * */
    private static <T> int count(MyStream<T> myStream, int count){
        if(myStream.isEmptyStream()){
            return count;
        }

        // count+1 进行递归
        return count(myStream.eval(),count+1);
    }

    /**
     * 递归函数 配合API.anyMatch
     * */
    private static <T> boolean anyMatch(Predicate<? super T> predicate,MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            // 截止末尾，不存在任何匹配项
            return false;
        }

        // 谓词判断
        if(predicate.satisfy(myStream.head)){
            // 匹配 存在匹配项 返回true
            return true;
        }else{
            // 不匹配，继续检查，直到存在匹配项
            return anyMatch(predicate,myStream.eval());
        }
    }

    /**
     * 递归函数 配合API.anyMatch
     * */
    private static <T> boolean allMatch(Predicate<? super T> predicate,MyStream<T> myStream){
        if(myStream.isEmptyStream()){
            // 全部匹配
            return true;
        }

        // 谓词判断
        if(predicate.satisfy(myStream.head)){
            // 当前项匹配，继续检查
            return allMatch(predicate,myStream.eval());
        }else{
            // 存在不匹配的项，返回false
            return false;
        }
    }

    /**
     * 当前流强制求值
     * @return 求值之后返回一个新的流
     * */
    private MyStream<T> eval(){
        return this.nextItemEvalProcess.eval();
    }

    /**
     * 当前流 为空
     * */
    private boolean isEmptyStream(){
        return this.isEnd;
    }

    @Override
    public String toString() {
        return "MyStream{" +
                "head=" + head +
                ", isEnd=" + isEnd +
                '}';
    }
}