package com.neverwinterdp.scribengin.dataflow.chain;

import java.util.ArrayList;
import java.util.List;

import com.neverwinterdp.scribengin.ScribenginClient;
import com.neverwinterdp.scribengin.dataflow.DataflowDescriptor;
import com.neverwinterdp.scribengin.dataflow.DataflowSubmitter;

public class OrderDataflowChainSubmitter extends DataflowChainSubmitter {
  private List<DataflowSubmitter> submitters = new ArrayList<>();
  
  public OrderDataflowChainSubmitter(ScribenginClient client, String dataflowHome, DataflowChainConfig config) {
    super(client, dataflowHome, config);
  }
  
  public void submit(long timeout) throws Exception {
    long stopTime = System.currentTimeMillis() + timeout;
    long remainTime = timeout;
    for(DataflowDescriptor sel : config.getDescriptors()) {
      DataflowSubmitter submitter = doSubmit(client, dataflowHome, sel) ;
      submitters.add(submitter);
      remainTime = stopTime - System.currentTimeMillis();
      if(remainTime < 0) {
        throw new InterruptedException("Cannot finish to submit within " + timeout + "ms") ;
      }
    }
  }
  
  protected DataflowSubmitter doSubmit(ScribenginClient client, String dataflowHome, DataflowDescriptor descriptor) throws Exception {
    DataflowSubmitter submitter = new DataflowSubmitter(client, dataflowHome, descriptor) ;
    submitter.submit();
    submitter.waitForRunning(30000);
    return submitter;
  }
  
  public void waitForTerminated(long timeout) throws Exception {
    long stopTime = System.currentTimeMillis() + timeout;
    long remainTime = timeout;
    for(DataflowSubmitter submitter : submitters) {
      submitter.waitForTerminated(remainTime);
      remainTime = stopTime - System.currentTimeMillis();
    }
  }
}
@@ -99,16 +99,56 @@
 .List;%0A%0A
+import com.neverwinterdp.registry.Node;%0A
 import c
@@ -321,16 +321,79 @@
 bmitter;
+%0Aimport com.neverwinterdp.scribengin.service.ScribenginService;
 %0A%0Apublic
@@ -1330,24 +1330,36 @@
 Exception %7B%0A
+    try %7B%0A  
     Dataflow
@@ -1434,24 +1434,26 @@
 scriptor) ;%0A
+  
     submitte
@@ -1464,24 +1464,26 @@
 bmit();%0A    
+  
 submitter.wa
@@ -1499,10 +1499,10 @@
 ing(
-30
+45
 000)
@@ -1507,16 +1507,18 @@
 0);%0A    
+ 
+ 
 return s
@@ -1527,16 +1527,212 @@
 mitter;%0A
+    %7D catch(Exception ex ) %7B%0A      Node dataflowNode = client.getRegistry().get(ScribenginService.getDataflowPath(descriptor.getId()));%0A      dataflowNode.dump(System.err);%0A      throw ex ;%0A    %7D%0A
   %7D%0A  %0A 

add error information