rafaelbernabeu / recursive-tasker Goto Github PK

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.function.Consumer;

public class RecursiveTasker<T> extends RecursiveTask<Collection<T>> {

    private final List<T> tasks;
    private final int start;
    private final int end;
    private final Consumer<T> itemAction;
    private ForkJoinPool pool;
    private int splitValue;

    public RecursiveTasker(Collection<T> tasks, Consumer<T> itemAction) {
        this(tasks, 0, tasks.size(), itemAction, getSplitValue(tasks), null);
    }

    public RecursiveTasker(Collection<T> tasks, Consumer<T> itemAction, ForkJoinPool pool) {
        this(tasks, 0, tasks.size(), itemAction, getSplitValue(tasks), pool);
    }

    private RecursiveTasker(Collection<T> tasks, int start, int end, Consumer<T> itemAction, Integer splitValue, ForkJoinPool pool) {
        this.pool = pool;
        this.tasks = new ArrayList<>(tasks);
        this.start = start;
        this.end = end;
        this.itemAction = itemAction;
        this.splitValue = splitValue;
    }

    private static <T> int getSplitValue(Collection<T> tasks) {
        return tasks.size() / (Runtime.getRuntime().availableProcessors() * 2);
    }

    @Override
    protected Collection<T> compute() {
        if (end - start <= splitValue) {
            for (int i = start; i < end; i++) {
                try {
                    itemAction.accept(tasks.get(i));
                } catch (Exception e) {
                    pool.shutdownNow();
                    throw e;
                }
            }
            return tasks.subList(start, end);
        } else {
            int middle = start + ((end - start) / 2);
            RecursiveTask<Collection<T>> otherTask = new RecursiveTasker<>(tasks, start, middle, itemAction, splitValue, pool);
            otherTask.fork();
            Collection<T> resultList = new RecursiveTasker<>(tasks, middle, end, itemAction, splitValue, pool).compute();
            resultList.addAll(otherTask.join());
            return resultList;
        }
    }

    public Collection<T> start(int splitValue) {
        this.splitValue = splitValue;
        return start();
    }

    public Collection<T> start() {
        if (pool == null) {
            pool = new ForkJoinPool();
        }
        return pool.invoke(this);
    }

}