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

import lombok.experimental.UtilityClass;

import net.ranides.assira.functional.Consumers;
import net.ranides.assira.functional.Predicates;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ForkJoinPool;
import java.util.function.Consumer;
import java.util.function.Predicate;

/**
 * This is internal utility class not supposed to use by clients.
 * It has default implementations of parallel query operations for arrays.
 *
 * Those methods should not be called directly, but by "BaseArray" facade.
 */
@UtilityClass
@SuppressWarnings("MissingJavadoc")
public class BaseArrayParalell {

    public static <T> void forEach(T[] array, Consumer<? super T> consumer) {
        forEach(array, 0, array.length, consumer);
    }

    public static <T> void forEach(T[] array, int begin, int end, Consumer<? super T> consumer) {
        Arrays.stream(array, begin, end).parallel().forEach(consumer);
    }

    public static <T> void forEach(T[] array, Consumers.EachConsumer<? super T> consumer) {
        forEach(array, 0, array.length, consumer);
    }

    public static <T> void forEach(T[] array, int begin, int end, Consumers.EachConsumer<? super T> consumer) {
        int length = end - begin;
        int threads = ForkJoinPool.getCommonPoolParallelism();
        int blocksize = Math.max(1, length / threads);
        List<Callable<Void>> tasks = new ArrayList<>(threads);

        int x = 0;
        for (int offset = begin, bound=end-blocksize; offset < bound; offset += blocksize) {
            int let_offset = offset;
            tasks.add(() -> {
                loop(array, begin, let_offset, let_offset + blocksize, consumer);
                return null;
            });
        }

        int tail = length % threads;
        if (tail != 0) {
            tasks.add(() -> {
                loop(array, begin, end - tail, end, consumer);
                return null;
            });
        }
        ForkJoinPool.commonPool().invokeAll(tasks);
    }

    public static <T> boolean whileEach(T[] array, Predicate<? super T> consumer) {
        return whileEach(array, 0, array.length, consumer);
    }

    public static <T> boolean whileEach(T[] array, int begin, int end, Predicate<? super T> consumer) {
        return Arrays.stream(array, begin, end).parallel().allMatch(consumer);
    }

    public static <T> boolean whileEach(T[] array, Predicates.EachPredicate<? super T> consumer) {
        return whileEach(array, 0, array.length, consumer);
    }

    public static <T> boolean whileEach(T[] array, int begin, int end, Predicates.EachPredicate<? super T> consumer) {
        int length = end - begin;
        int threads = ForkJoinPool.getCommonPoolParallelism();
        int blocksize = Math.max(1, length / threads);
        List<Callable<Void>> tasks = new ArrayList<>(threads);
        BaseState.Terminator terminator = BaseState.newTerminator(true);

        for (int offset = begin; offset < end; offset += blocksize) {
            int let_offset = offset;
            tasks.add(() -> {
                loop(array, begin, let_offset, let_offset+blocksize, terminator, consumer);
                return null;
            });
        }

        int tail = length % threads;
        if (tail != 0) {
            tasks.add(() -> {
                loop(array, begin, end - tail, end, terminator, consumer);
                return null;
            });
        }

        ForkJoinPool.commonPool().invokeAll(tasks);
        return !terminator.stopped();
    }

    public static <T> void loop(T[] array, int offset, int begin, int end, Consumers.EachConsumer<? super T> consumer) {
        for (int index = begin; index < end; index++) {
            consumer.accept(index - offset, array[index]);
        }
    }

    public static <T> void loop(T[] array, int offset, int begin, int end, BaseState.Terminator terminator, Predicates.EachPredicate<? super T> consumer) {
        for (int index = begin; index < end; index++) {
            if(terminator.stopped()) {
                break;
            }
            boolean test = consumer.test(index - offset, array[index]);
            if(!test) {
                terminator.stop();
                break;
            }
        }
    }


}
