package com.breakersoft.plow.dispatcher;

import java.util.UUID;

import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.breakersoft.plow.Job;
import com.breakersoft.plow.Layer;
import com.breakersoft.plow.LayerE;
import com.breakersoft.plow.Task;
import com.breakersoft.plow.dispatcher.domain.DispatchProc;
import com.breakersoft.plow.rnd.thrift.RunTaskResult;
import com.breakersoft.plow.service.DependService;
import com.breakersoft.plow.service.JobService;
import com.breakersoft.plow.service.StateManager;
import com.breakersoft.plow.thrift.TaskState;

@Component
public class TaskCompleteHandler {

    private static final Logger logger =
            org.slf4j.LoggerFactory.getLogger(TaskCompleteHandler.class);

    @Autowired
    DispatchService dispatchService;

    @Autowired
    DependService dependService;

    @Autowired
    JobService jobService;

    @Autowired
    StateManager jobStateManager;

    @Autowired
    ProcDispatcher procDispatcher;

    public void taskComplete(RunTaskResult result) {

        Task task = jobService.getTask(UUID.fromString(result.taskId));
        Job job = jobService.getJob(UUID.fromString(result.jobId));
        DispatchProc proc = dispatchService.getDispatchProc(result.procId);

        TaskState newState;
        if (result.exitStatus == 0) {
            newState = TaskState.SUCCEEDED;
        }
        else {
            newState = TaskState.DEAD;
        }

        logger.info("New state {}", newState.toString());

        if (dispatchService.stopTask(task, newState)) {
            dispatchService.unassignProc(proc);

            if (newState.equals(TaskState.SUCCEEDED)) {
                jobStateManager.satisfyDependsOn(task);
                final Layer layer = new LayerE(task);
                if (jobService.isLayerComplete(layer)) {
                    jobStateManager.satisfyDependsOn(layer);
                }
            }
        }

        if (proc.isUnbooked()) {
            dispatchService.deallocateProc(proc, "Task was stopped");
        }

        if (jobService.isJobPaused(job)) {
            dispatchService.deallocateProc(proc,
                    "Job is paused: " + job.getJobId());
            return;
        }

        if (jobService.isFinished(job)) {
            dispatchService.deallocateProc(proc,
                    "Job is finished " + job.getJobId());
            jobStateManager.shutdownJob(job);
            return;
        }

        procDispatcher.book(proc);
    }
}

@@ -147,16 +147,79 @@
 owired;%0A
+import org.springframework.dao.EmptyResultDataAccessException;%0A
 import o
@@ -1190,16 +1190,113 @@
 k task =
+ null;%0A        Job job = null;%0A        DispatchProc proc = null;%0A        try %7B%0A            task =
  jobServ
@@ -1344,27 +1344,27 @@
 ));%0A        
+  
-Job
+ 
  job = jobSe
@@ -1412,36 +1412,27 @@
 ));%0A        
-DispatchProc
+   
  proc = disp
@@ -1474,16 +1474,211 @@
 procId);
+%0A        %7D%0A        catch (EmptyResultDataAccessException e) %7B%0A            logger.error(%22Bad task complete ping Job:%22 + result.jobId + %22, Task: %22 + result.taskId, e);%0A            return;%0A        %7D
 %0A%0A      

Handle case where proc pings in something already cleaned up by orphan killer