package com.neverwinterdp.scribengin.dataflow.worker;

import com.neverwinterdp.registry.RegistryException;
import com.neverwinterdp.registry.task.TaskContext;
import com.neverwinterdp.scribengin.dataflow.DataflowDescriptor;
import com.neverwinterdp.scribengin.dataflow.DataflowRegistry;
import com.neverwinterdp.scribengin.dataflow.DataflowTask;
import com.neverwinterdp.scribengin.dataflow.DataflowTaskDescriptor;
import com.neverwinterdp.yara.MetricRegistry;
import com.neverwinterdp.yara.Timer;

public class DataflowTaskExecutor {
  private DataflowTaskExecutorDescriptor executorDescriptor;
  private ExecutorManagerThread          executorManagerThread;
  private DataflowTaskExecutorThread     executorThread;
  private DataflowTaskExecutorService    executorService;
  private DataflowTask                   currentDataflowTask = null;
  private boolean                        interrupt           = false;
  private boolean                        kill                = false;

  public DataflowTaskExecutor(DataflowTaskExecutorService service, DataflowTaskExecutorDescriptor  descriptor) throws RegistryException {
    executorDescriptor = descriptor;
    this.executorService = service;
    service.getDataflowRegistry().createWorkerTaskExecutor(service.getVMDescriptor(), descriptor);
  }
  
  public DataflowTaskExecutorDescriptor getDescriptor() { return this.executorDescriptor ; }
  
  public void start() {
    interrupt = false ;
    executorManagerThread = new ExecutorManagerThread();
    executorManagerThread.start();
  }
  
  public void interrupt() throws Exception {
    if(isAlive()) {
      interrupt = true ;
      if(currentDataflowTask != null) currentDataflowTask.interrupt();
    }
  }
  
  public boolean isAlive() {
    if(executorManagerThread == null) return false;
    return executorManagerThread.isAlive();
  }
  
  public void execute() { 
    executorDescriptor.setStatus(DataflowTaskExecutorDescriptor.Status.RUNNING);
    MetricRegistry metricRegistry = executorService.getMetricRegistry();
    DataflowRegistry dataflowRegistry = executorService.getDataflowRegistry();
    Timer dataflowTaskTimerGrab = metricRegistry.getTimer("dataflow-task.timer.grab") ;
    Timer dataflowTaskTimerProcess = metricRegistry.getTimer("dataflow-task.timer.process") ;
    try {
      DataflowDescriptor dflDescriptor = dataflowRegistry.getDataflowDescriptor(false);
      while(!interrupt) {
        TaskContext<DataflowTaskDescriptor> taskContext = null ;
        int retries = 0 ;
        while(taskContext == null && retries < 3) {
          Timer.Context dataflowTaskTimerGrabCtx = dataflowTaskTimerGrab.time() ;
          taskContext= dataflowRegistry.dataflowTaskAssign(executorService.getVMDescriptor());
          dataflowTaskTimerGrabCtx.stop();
          Thread.sleep(3000);
          retries++;
        }
        

        if(taskContext == null) return;
        if(interrupt) {
          dataflowRegistry.dataflowTaskSuspend(taskContext);
          return ;
        }
        
        Timer.Context dataflowTaskTimerProcessCtx = dataflowTaskTimerProcess.time() ;
        executorDescriptor.addAssignedTask(taskContext.getTaskTransactionId().getTaskId());
        dataflowRegistry.updateWorkerTaskExecutor(executorService.getVMDescriptor(), executorDescriptor);
        currentDataflowTask = new DataflowTask(executorService, taskContext);
        currentDataflowTask.init();
        executorThread = new DataflowTaskExecutorThread(currentDataflowTask);
        executorThread.start();
        executorThread.waitForTimeout(dflDescriptor.getTaskSwitchingPeriod());
        if(currentDataflowTask.isComplete()) {
          currentDataflowTask.finish();
        } else {
          currentDataflowTask.suspend();
        }
        dataflowTaskTimerProcessCtx.stop();
      }
      doExit(DataflowTaskExecutorDescriptor.Status.TERMINATED);
    } catch (InterruptedException e) {
      executorService.getLogger().error("DataflowTaskExecutor: detect shutdown interrupt for task " + currentDataflowTask.getDescriptor().getTaskId());
      currentDataflowTask.interrupt();
      doExit(DataflowTaskExecutorDescriptor.Status.TERMINATED_WITH_INTERRUPT);
    } catch (Throwable e) {
      executorService.getLogger().error("DataflowTaskExecutor Error", e);
      doExit(DataflowTaskExecutorDescriptor.Status.TERMINATED_WITH_ERROR);
    }
  }

  void doExit(DataflowTaskExecutorDescriptor.Status status) {
    if(kill) return ;
    try {
      executorDescriptor.setStatus(status);
      DataflowRegistry dataflowRegistry = executorService.getDataflowRegistry();
      dataflowRegistry.updateWorkerTaskExecutor(executorService.getVMDescriptor(), executorDescriptor);
    } catch(Exception ex) {
      ex.printStackTrace();
    }
  }
  
  /**
   * This method is used to simulate the failure
   * @throws Exception
   */
  public void kill() throws Exception {
    kill = true;
    if(executorThread != null && executorThread.isAlive()) executorThread.interrupt();
    if(executorManagerThread != null && executorManagerThread.isAlive()) executorManagerThread.interrupt();
  }
  
  public class ExecutorManagerThread extends Thread {
    public void run() {
      execute();
    }
  }
  
  public class DataflowTaskExecutorThread extends Thread {
    DataflowTask  dataflowtask;
    private boolean terminated = false;
    
    public DataflowTaskExecutorThread(DataflowTask  dataflowtask) {
      this.dataflowtask = dataflowtask;
    }

    public void run() {
      dataflowtask.execute();
      notifyTermination();
    }
    
    synchronized public void notifyTermination() {
      terminated = true;
      notifyAll() ;
    }
    
    synchronized void waitForTimeout(long timeout) throws InterruptedException {
      if(timeout > 0) wait(timeout);
      else wait();
      if(!terminated) dataflowtask.interrupt();
      waitForTerminated();
    }
    
    synchronized void waitForTerminated() throws InterruptedException {
      if(terminated) return ;
      wait(3000);
    }
  }
}
@@ -2415,153 +2415,8 @@
 ) %7B%0A
-        TaskContext%3CDataflowTaskDescriptor%3E taskContext = null ;%0A        int retries = 0 ;%0A        while(taskContext == null && retries %3C 3) %7B%0A  
     
@@ -2503,22 +2503,70 @@
     
-  taskContext=
+TaskContext%3CDataflowTaskDescriptor%3E taskContext = %0A           
  dat
@@ -2634,34 +2634,32 @@
 tor());%0A        
-  
 dataflowTaskTime
@@ -2675,24 +2675,25 @@
 stop();%0A
+%0A
         
   Thread
@@ -2688,27 +2688,165 @@
     
-  Thread.sleep(3000
+if(interrupt) %7B%0A          dataflowRegistry.dataflowTaskSuspend(taskContext);%0A          doExit(DataflowTaskExecutorDescriptor.Status.TERMINATED_WITH_INTERRUPT
 );%0A 
@@ -2861,14 +2861,12 @@
  ret
-ries++
+urn 
 ;%0A  
@@ -2882,17 +2882,16 @@
        %0A
-%0A
         
@@ -2918,98 +2918,75 @@
 ll) 
-return;%0A        if(interrupt) %7B%0A          dataflowRegistry.dataflowTaskSuspend(taskContext
+%7B%0A          doExit(DataflowTaskExecutorDescriptor.Status.TERMINATED
 );%0A 
@@ -2992,33 +2992,32 @@
           return
- 
 ;%0A        %7D%0A    
@@ -3819,302 +3819,8 @@
   %7D%0A
-      doExit(DataflowTaskExecutorDescriptor.Status.TERMINATED);%0A    %7D catch (InterruptedException e) %7B%0A      executorService.getLogger().error(%22DataflowTaskExecutor: detect shutdown interrupt for task %22 + currentDataflowTask.getDescriptor().getTaskId());%0A      currentDataflowTask.interrupt();%0A
     
@@ -4445,25 +4445,87 @@
   ex
-.printStackTrace(
+ecutorService.getLogger().error(%22DataflowTaskExecutor Fail To Updat Status%22, ex
 );%0A 

fix status report for task executor