package be.iminds.ilabt.jfed.highlevel.jobs;

import be.iminds.ilabt.jfed.experiment.Experiment;
import be.iminds.ilabt.jfed.experiment.tasks.TaskExecutionFinishedListener;
import be.iminds.ilabt.jfed.highlevel.controller.Task;
import be.iminds.ilabt.jfed.highlevel.controller.TaskExecution;
import be.iminds.ilabt.jfed.highlevel.controller.TaskExecutionFinishedCallback;
import be.iminds.ilabt.jfed.highlevel.controller.TaskThread;
import be.iminds.ilabt.jfed.highlevel.jobs.report.JobReport;
import be.iminds.ilabt.jfed.highlevel.tasks.HighLevelTaskFactory;
import be.iminds.ilabt.jfed.lowlevel.JFedException;
import javafx.application.Platform;
import javafx.beans.property.ReadOnlyObjectProperty;
import javafx.beans.property.ReadOnlyObjectWrapper;
import javafx.beans.property.ReadOnlyStringProperty;
import javafx.beans.property.ReadOnlyStringWrapper;
import javafx.collections.FXCollections;
import javafx.collections.ObservableList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;

import javax.annotation.Nonnull;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * User: twalcari
 * Date: 7/13/2015
 * Time: 16:30
 */
public abstract class AbstractJob<V> implements Job<V> {
    private static final Logger LOG = LoggerFactory.getLogger(AbstractJob.class);

    private final Marker jobMarker;

    private final String name;
    protected final Experiment experiment;

    protected final HighLevelTaskFactory hltf;
    private final TaskThread tt;

    private final ReadOnlyObjectWrapper<State> state = new ReadOnlyObjectWrapper<>();
    private final ReadOnlyStringWrapper message = new ReadOnlyStringWrapper();
    private final ObservableList<TaskExecution> taskExecutions = FXCollections.observableArrayList();
    private final List<TaskExecutionFinishedListener> taskExecutionFinishedListeners = new ArrayList<>();

    private final JobReport jobReport;


    public AbstractJob(String name, Experiment experiment, HighLevelTaskFactory hltf, TaskThread tt) {
        super();
        this.name = name;
        this.experiment = experiment;
        this.hltf = hltf;
        this.tt = tt;

        this.jobReport = new JobReport(this);

        this.jobMarker = MarkerFactory.getMarker(name);
    }

    @Nonnull
    public Collection<Task> getAlwaysDependsOn() {
        return Collections.emptyList();
    }

    @Nonnull
    public Collection<Task> getDependsOn() {
        return Collections.emptyList();
    }

    @Override
    public final V call() throws Exception {
        //first check and execute any dependencies that we have

        List<Task> tasksToExecute = new ArrayList<>();

        tasksToExecute.addAll(getAlwaysDependsOn());
        tasksToExecute.addAll(getDependsOn().stream()
                .filter(Task::newExecutionNeeded)
                .collect(Collectors.toList()));

        if (!tasksToExecute.isEmpty()) {
            CountDownLatch latch = new CountDownLatch(tasksToExecute.size());

            List<TaskExecution> taskExecutions = new ArrayList<>();

            for (Task task : tasksToExecute) {
                TaskExecution taskExecution = submitTask(task, (t, te, state) -> latch.countDown());
                taskExecutions.add(taskExecution);
            }

            latch.await();

            //check if everything was successful

            for (TaskExecution taskExecution : taskExecutions) {
                if (taskExecution.getState() != TaskExecution.TaskState.SUCCESS) {
                    LOG.error(jobMarker, "Dependency {} has state {} instead of SUCCESS. Aborting this job!",
                            taskExecution.getName(), taskExecution.getState());
                    return null;
                }
            }
        }


        //now run the real task
        return execute();
    }

    public abstract V execute() throws Exception;

    private <T extends Task> TaskExecutionFinishedCallback<T> createTaskExecutionFinishedCallbackForTask() {
        return (t, te, s) -> taskExecutionFinishedListeners.forEach(listener -> listener.onTaskExecutionFinished(te));
    }

    @Override
    public <T extends Task> TaskExecution<T> submitTask(T task, TaskExecutionFinishedCallback<T> callback) {
        List<TaskExecutionFinishedCallback<T>> callbacks = new ArrayList<>();
        callbacks.add(callback);
        callbacks.add(createTaskExecutionFinishedCallbackForTask());

        TaskExecution<T> taskExecution = tt.addTask(task, callbacks);
        taskExecutions.add(taskExecution);

        return taskExecution;
    }

    @Override
    public <T extends Task> TaskExecution<T> submitTask(T task) {
        TaskExecution<T> taskExecution = tt.addTask(task, createTaskExecutionFinishedCallbackForTask());
        taskExecutions.add(taskExecution);

        return taskExecution;
    }

    @Override
    public <T extends Task> TaskExecution<T> submitTaskAndWait(T task) throws InterruptedException {
        CountDownLatch latch = new CountDownLatch(1);

        TaskExecution<T> taskExecution = submitTask(task, (t, te, state) -> latch.countDown());

        latch.await();

        return taskExecution;
    }

    protected void setAndRunState(State newState) throws JFedException {
        this.state.set(newState);
        newState.run(this);
    }


    @Override
    public State getState() {
        return state.get();
    }

    @Override
    public final ReadOnlyObjectProperty<State> stateProperty() {
        return state.getReadOnlyProperty();
    }

    protected void updateMessage(String message) {

        Platform.runLater(() -> this.message.set(message));
    }

    @Override
    public final ReadOnlyStringProperty messageProperty() {
        return message.getReadOnlyProperty();
    }

    @Override
    public final String getMessage() {
        return message.get();
    }

    @Override
    public final String getName() {
        return name;
    }

    @Nonnull
    public ObservableList<TaskExecution> getTaskExecutions() {
        return /*FXCollections.unmodifiableObservableList(*/taskExecutions/*)*/;
    }

    public void addTaskExecutionFinishedListener(TaskExecutionFinishedListener taskExecutionFinishedListener) {
        taskExecutionFinishedListeners.add(taskExecutionFinishedListener);
    }

    @Override
    public Experiment getExperiment() {
        return experiment;
    }
}
