package com.neverwinterdp.scribengin.dataflow.tracking;

import com.beust.jcommander.Parameter;
import com.neverwinterdp.registry.txevent.TXEvent;
import com.neverwinterdp.scribengin.ScribenginClient;
import com.neverwinterdp.scribengin.dataflow.Dataflow;
import com.neverwinterdp.scribengin.dataflow.DataflowClient;
import com.neverwinterdp.scribengin.dataflow.DataflowEvent;
import com.neverwinterdp.scribengin.shell.ScribenginShell;
import com.neverwinterdp.vm.client.VMClient;

public class TrackingWithSimulationLauncher extends TrackingLauncher {
  @Parameter(names = "--simulation-report-period", description = "")
  long simulationReportPeriod = 5000;
  
  @Parameter(names = "--simulation-period", description = "")
  long simulationPeriod = 30000;
  
  @Parameter(names = "--simulation-max", description = "")
  int simulationMax = 2;
  
  private int stopCount  = 0;
  private int startCount = 0;
  
  @Override
  public void execute(ScribenginShell shell, TrackingDataflowBuilder dflBuilder) throws Exception {
    VMClient vmClient = shell.getVMClient();
    submitVMGenerator(vmClient, dflBuilder);
    
    Dataflow<TrackingMessage, TrackingMessage> dfl = dflBuilder.buildDataflow();
    dfl.setDefaultParallelism(5);
    dfl.setDefaultReplication(1);
    startDataflow(shell, dfl);
    
    submitVMValidator(vmClient, dflBuilder);
    String reportPath = dflBuilder.getTrackingConfig().getReportPath();
    for(int i = 0; i < simulationMax; i++) {
      long stopTime = System.currentTimeMillis() + simulationPeriod;
      long reportTime = 0;
      while(System.currentTimeMillis() < stopTime) {
        Thread.sleep(1000);
        reportTime +=  1000;
        if(reportTime >= simulationReportPeriod) {
          shell.execute("dataflow info" + "  --dataflow-id " + dfl.getDataflowId() + " --show-history-workers");
          shell.execute(
              "plugin com.neverwinterdp.scribengin.dataflow.tracking.TrackingMonitor" +
              "  --dataflow-id " + dfl.getDataflowId() + " --show-history-workers  --report-path " + reportPath
          );
          reportTime = 0;
        }
      }
      stopDataflow(shell, dflBuilder);
      submitDataflow(shell, dfl);
    }
  }
  
  public void startDataflow(ScribenginShell shell, Dataflow<TrackingMessage, TrackingMessage> dfl) throws Exception {
    System.err.println("--------------------------------------------------------------------------------------");
    System.err.println("Start Dataflow, stopCount = " + startCount);
    System.err.println("--------------------------------------------------------------------------------------");
    submitDataflow(shell, dfl);
    startCount++;
  }
  
  public void stopDataflow(ScribenginShell shell,TrackingDataflowBuilder dflBuilder) throws Exception {
    System.err.println("--------------------------------------------------------------------------------------");
    System.err.println("Stop Dataflow, stopCount = " + stopCount);
    System.err.println("--------------------------------------------------------------------------------------");
    ScribenginShell scribenginShell = (ScribenginShell) shell;
    ScribenginClient scribenginClient = scribenginShell.getScribenginClient();
    String dataflowId = dflBuilder.getDataflowId();
    String reportPath = dflBuilder.getTrackingConfig().getReportPath();
    DataflowClient dflClient = scribenginClient.getDataflowClient(dataflowId);
    TXEvent stopEvent = new TXEvent("stop", DataflowEvent.Stop);
    dflClient.getDataflowRegistry().getMasterRegistry().getMaserEventBroadcaster().broadcast(stopEvent);
    shell.execute("dataflow wait-for-status --dataflow-id "  + dataflowId + " --status TERMINATED --timeout 60000") ;
    shell.execute(
      "plugin com.neverwinterdp.scribengin.dataflow.tracking.TrackingMonitor" +
      "  --dataflow-id " + dataflowId + " --show-history-workers  --report-path " + reportPath
    );
    stopCount++ ;
  }
}

@@ -89,16 +89,130 @@
 ameter;%0A
+import com.neverwinterdp.message.MessageTrackingRegistry;%0Aimport com.neverwinterdp.message.MessageTrackingReport;%0A
 import c
@@ -3705,120 +3705,662 @@
    s
-hell.execute(%22dataflow wait-for-status --dataflow-id %22  + dataflowId + %22 --status TERMINATED --time
+topCount++ ;%0A    %0A    MessageTrackingRegistry mtRegistry = dflClient.getDataflowRegistry().getMessageTrackingRegistry();%0A    while(true) %7B%0A      MessageTrackingReport inputReporter  = mtRegistry.getMessageTrackingReporter(%22input%22);%0A      MessageTrackingReport outputReporter = mtRegistry.getMessageTrackingReporter(%22output%22);%0A      long inputCount  = inputReporter.getTrackingCount();%0A      long outputCount = outputReporter.getTrackingCount();%0A      System.err.println(%22Stop: input count = %22 + inputCount + %22, output count = %22 + outputCount);%0A      if(inputC
 ou
+n
 t 
-60000%22) ;
+== outputCount )%7B%0A        break;%0A      %7D%0A      Thread.sleep(simulationReportPeriod);%0A    %7D
 %0A   
@@ -4371,24 +4371,26 @@
 ll.execute(%0A
+  
       %22plugi
@@ -4453,24 +4453,26 @@
 gMonitor%22 +%0A
+  
       %22  --d
@@ -4562,31 +4562,131 @@
     );%0A    s
-topCount++
+hell.execute(%22dataflow wait-for-status --dataflow-id %22  + dataflowId + %22 --status TERMINATED --timeout 90000%22)
  ;%0A  %7D%0A%7D%0A

Implement start/stop/resume. Bug fix and code enhancement to support start/stop/resume