package com.neverwinterdp.scribengin.dataflow.operator;

import com.neverwinterdp.registry.task.TaskStatus;
import com.neverwinterdp.registry.task.dedicated.DedicatedTaskContext;
import com.neverwinterdp.registry.task.dedicated.TaskSlotExecutor;
import com.neverwinterdp.scribengin.dataflow.registry.DataflowRegistry;
import com.neverwinterdp.scribengin.dataflow.worker.WorkerService;
import com.neverwinterdp.scribengin.storage.Record;

public class OperatorTaskSlotExecutor extends TaskSlotExecutor<OperatorTaskConfig>{
  private WorkerService                            workerService;
  private DedicatedTaskContext<OperatorTaskConfig> taskContext;
  private OperatorTaskConfig                       operatorTaskConfig;
  private Operator                                 operator;
  private OperatorContext                          context;
  
  private long                                     startTime = 0;

  public OperatorTaskSlotExecutor(WorkerService service, DedicatedTaskContext<OperatorTaskConfig> taskContext) throws Exception {
    super(taskContext);
    this.workerService = service;
    this.taskContext     = taskContext;
    this.operatorTaskConfig      = taskContext.getTaskDescriptor(false);
    Class<Operator> opType = (Class<Operator>) Class.forName(operatorTaskConfig.getOperator());
    operator = opType.newInstance();
    
    startTime = System.currentTimeMillis();
    DataflowRegistry dRegistry = workerService.getDataflowRegistry();
    OperatorTaskReport report = dRegistry.getTaskRegistry().getTaskReport(operatorTaskConfig);
    report.incrAssignedCount();
    dRegistry.getTaskRegistry().save(operatorTaskConfig, report);
    context = new OperatorContext(workerService, operatorTaskConfig, report);
    dRegistry.getTaskRegistry().save(operatorTaskConfig, report);
  }
  
  public DedicatedTaskContext<OperatorTaskConfig> getTaskContext() { return this.taskContext; }
  
  public OperatorTaskConfig getDescriptor() { return operatorTaskConfig ; }
  
  public boolean isComplete() { return context.isComplete() ; }
  
  
  @Override
  public void executeSlot() throws Exception {
    OperatorTaskReport report = context.getTaskReport();
    int recCount = 0;
    try {
      while(!isInterrupted() && recCount <= 1000 && !context.isComplete()) {
        Record record = context.nextRecord(1000);
        if(record == null) break ;

        recCount++;
        report.incrProcessCount();
        operator.process(context, record);
      } //end while
      
      if(recCount == 0) {
        report.setAssignedWithNoMessageProcess(report.getAssignedWithNoMessageProcess() + 1);
        report.setLastAssignedWithNoMessageProcess(report.getLastAssignedWithNoMessageProcess() + 1);
      } else {
        report.setLastAssignedWithNoMessageProcess(0);
      }
      if(report.getLastAssignedWithNoMessageProcess() >= 3) {
        getTaskContext().setComplete();
        context.setComplete();
      }
      
      updateContext();
    } catch(InterruptedException ex) {
      //kill simulation
      throw ex ;
    } catch(Throwable t) {
      report.setAssignedHasErrorCount(report.getAssignedHasErrorCount() + 1);
      workerService.getLogger().error("DataflowTask Error", t);
      t.printStackTrace();
    }
  }
  
  public void suspend() throws Exception {
    saveContext();
    DataflowRegistry dflRegistry = workerService.getDataflowRegistry();
    dflRegistry.getTaskRegistry().suspend(taskContext);
  }
  
  public void finish() throws Exception {
    OperatorTaskReport report = context.getTaskReport();
    report.setFinishTime(System.currentTimeMillis());
    saveContext();
    DataflowRegistry dflRegistry = workerService.getDataflowRegistry();
    dflRegistry.getTaskRegistry().finish(taskContext, TaskStatus.TERMINATED);
  }
  
  void saveContext() {
    try {
      OperatorTaskReport report = context.getTaskReport();
      report.addAccRuntime(System.currentTimeMillis() - startTime);
      context.commit();
      context.close();
    } catch(Exception ex) {
      workerService.getLogger().error("Cannot save the executor context due to the error: " + ex);
    }
  }
  
  void updateContext() throws Exception {
    OperatorTaskReport report = context.getTaskReport();
    report.addAccRuntime(System.currentTimeMillis() - startTime);
    context.commit();
  }
}
@@ -906,16 +906,184 @@
 me = 0;%0A
+  private long                                     lastFlushTime = System.currentTimeMillis();%0A  private long                                     lastNoMessageTime ;%0A  
 %0A  publi
@@ -2217,19 +2217,16 @@
  ; %7D%0A  %0A
-  %0A
   @Overr
@@ -2663,27 +2663,147 @@
     
-if(recCount == 0) %7B
+long currentTime = System.currentTimeMillis();%0A      if(recCount == 0) %7B%0A        if(lastNoMessageTime %3C 0) lastNoMessageTime = currentTime;
 %0A   
@@ -3005,72 +3005,133 @@
     
-%7D else %7B%0A        report.setLastAssignedWithNoMessageProcess(0
+  if(lastNoMessageTime + 15000 %3C currentTime) %7B%0A          getTaskContext().setComplete();%0A          context.setComplete(
 );%0A
+  
     
@@ -3144,19 +3144,33 @@
     
-if(
+%7D else %7B%0A        
 report.
-g
+s
 etLa
@@ -3204,74 +3204,76 @@
 ess(
-) %3E= 3) %7B%0A        getTaskContext().setComplete()
+0);%0A        lastNoMessageTime = -1
 ;%0A     
+ 
-   
+%7D%0A      %0A      if(
 context.
 setC
@@ -3260,35 +3260,34 @@
      if(context.
+i
 s
-et
 Complete();%0A    
@@ -3280,33 +3280,96 @@
 mplete()
-;%0A      %7D%0A      %0A
+ %7C%7C report.getProcessCount() %3E 5000 %7C%7C lastFlushTime + 10000 %3C currentTime) %7B%0A  
       up
@@ -3379,24 +3379,32 @@
 eContext();%0A
+      %7D%0A
     %7D catch(

change end of task detection