package mtsar.processors.task;

import mtsar.api.Process;
import mtsar.api.TaskAllocation;
import mtsar.api.Worker;
import mtsar.api.sql.AnswerDAO;
import mtsar.api.sql.TaskDAO;
import mtsar.processors.TaskAllocator;

import javax.inject.Inject;
import javax.inject.Provider;
import java.util.Optional;

public class FixedNumberAllocator implements TaskAllocator {
    protected final Provider<Process> process;
    protected final TaskDAO taskDAO;
    protected final AnswerDAO answerDAO;
    protected int answersPerTask;

    @Inject
    public FixedNumberAllocator(Provider<Process> processProvider, TaskDAO taskDAO, AnswerDAO answerDAO) {
        this.process = processProvider;
        this.taskDAO = taskDAO;
        this.answerDAO = answerDAO;
    }

    @Override
    public Optional<TaskAllocation> allocate(Worker w) {
        ensureAnswersPerTask();
        throw new UnsupportedOperationException("Not Implemented Yet");
    }

    protected void ensureAnswersPerTask() {
        final Integer answersPerTask = Integer.valueOf(process.get().getOptions().get("answersPerTask"));
        if (answersPerTask == null) {
            throw new RuntimeException("answersPerTask option is not set");
        }
        this.answersPerTask = answersPerTask;
    }
}

package mtsar.processors.task;

import mtsar.api.Process;
import mtsar.api.Task;
import mtsar.api.Worker;
import mtsar.api.sql.AnswerDAO;
import mtsar.api.sql.TaskDAO;

import javax.inject.Inject;
import javax.inject.Provider;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;

public class FixedNumberAllocator extends InverseCountAllocator {
    private Integer answersPerTask = null;

    @Inject
    public FixedNumberAllocator(Provider<Process> processProvider, TaskDAO taskDAO, AnswerDAO answerDAO) {
        super(processProvider, taskDAO, answerDAO);
    }

    @Override
    protected Map<Task, Integer> getCounts(List<Task> tasks, Worker worker) {
        ensureAnswersPerTask();

        final Map<Task, Boolean> answers = tasks.parallelStream().collect(
                Collectors.toMap(Function.identity(), t -> answerDAO.listForTask(t.getId(), process.get().getId()).stream().anyMatch(a -> a.getWorkerId().equals(worker.getId())))
        );

        final Map<Task, Integer> counts = super.getCounts(tasks, worker);

        final Iterator<Map.Entry<Task, Integer>> iterator = counts.entrySet().iterator();
        while (iterator.hasNext()) {
            final Map.Entry<Task, Integer> entry = iterator.next();
            if (answers.get(entry.getKey()) || (entry.getValue() >= answersPerTask)) iterator.remove();
        }

        return counts;
    }

    private void ensureAnswersPerTask() {
        final Integer answersPerTask = Integer.valueOf(process.get().getOptions().get("answersPerTask"));
        if (answersPerTask == null) throw new RuntimeException("answersPerTask option is not set");
        this.answersPerTask = answersPerTask;
    }
}

Implement FixedNumberAllocator