package com.neverwinterdp.dataflow.logsample;

import com.beust.jcommander.JCommander;
import com.neverwinterdp.dataflow.logsample.vm.VMLogMessageGeneratorApp;
import com.neverwinterdp.dataflow.logsample.vm.VMLogMessageValidatorApp;
import com.neverwinterdp.registry.Registry;
import com.neverwinterdp.scribengin.client.shell.ScribenginShell;
import com.neverwinterdp.scribengin.dataflow.chain.DataflowChainConfig;
import com.neverwinterdp.scribengin.dataflow.chain.OrderDataflowChainSubmitter;
import com.neverwinterdp.util.JSONSerializer;
import com.neverwinterdp.util.io.IOUtil;
import com.neverwinterdp.vm.VMConfig;
import com.neverwinterdp.vm.client.GroupVMSubmitter;
import com.neverwinterdp.vm.client.VMClient;
import com.neverwinterdp.vm.client.VMSubmitter;

public class LogSampleRunner {
  LogSampleConfig config ;
  ScribenginShell shell ;
  
  public LogSampleRunner(String[] args) throws Exception {
    config = new LogSampleConfig() ;
    new JCommander(config, args);
    Registry registry = config.registryConfig.newInstance();
    registry.connect();
    VMClient vmClient = new VMClient(registry);
    shell = new ScribenginShell(vmClient);
  }
  
  public void uploadApp() throws Exception {
    if(config.uploadApp != null) {
      shell.getVMClient().uploadApp(config.uploadApp, config.dfsAppHome);
    }
  }
  
  public void submitVMLogGeneratorApp() throws Exception {
    GroupVMSubmitter groupVMSubmitter = new GroupVMSubmitter(shell.getVMClient());
    for(int i = 0; i < config.logGeneratorNumOfVM; i++) {
      VMConfig vmConfig = new VMConfig();
      vmConfig.setRegistryConfig(config.registryConfig);
      vmConfig.setName("vm-log-generator-" + (i + 1));
      vmConfig.addRoles("vm-log-generator");
      vmConfig.setVmApplication(VMLogMessageGeneratorApp.class.getName());
      vmConfig.addProperty("num-of-executor", config.logGeneratorNumOfExecutorPerVm);
      vmConfig.addProperty("num-of-message-per-executor", config.logGeneratorNumOfMessagePerExecutor);
      vmConfig.addProperty("message-size", config.logGeneratorMessageSize);
      groupVMSubmitter.add(config.dfsAppHome, vmConfig);
    }
    groupVMSubmitter.submitAndWaitForRunning(45000);
  }
  
  public VMSubmitter submitVMLogValidatorApp() throws Exception {
    VMConfig vmConfig = new VMConfig() ;
    vmConfig.setRegistryConfig(config.registryConfig);
    vmConfig.setName("log-validator");
    vmConfig.addRoles("log-validator");
    vmConfig.addProperty("num-of-executor", config.logValidatorNumOfExecutorPerVM);
    vmConfig.addProperty("wait-for-message-timeout", config.logValidatorWaitForMessageTimeout);
    vmConfig.addProperty("wait-for-termination", config.logValidatorWaitForTermination);
    vmConfig.addProperty("validate-topic", config.logValidatorValidateTopic);
    vmConfig.setVmApplication(VMLogMessageValidatorApp.class.getName());
    VMSubmitter vmSubmitter = new VMSubmitter(shell.getVMClient(), config.dfsAppHome, vmConfig);
    vmSubmitter.submit();
    vmSubmitter.waitForRunning(30000);
    return vmSubmitter ;
  }
  
  public void submitLogSampleDataflowChain() throws Exception {
    String json = IOUtil.getFileContentAsString(config.dataflowDescriptor) ;
    DataflowChainConfig dflChainconfig = JSONSerializer.INSTANCE.fromString(json, DataflowChainConfig.class);
    OrderDataflowChainSubmitter submitter = 
        new OrderDataflowChainSubmitter(shell.getScribenginClient(), null, dflChainconfig);
    if(config.dataflowTaskDebug) {
      submitter.enableDataflowTaskDebugger();
    }
    submitter.submit(45000);
    submitter.waitForTerminated(45000);
  }
  
  static public void main(String[] args) throws Exception {
    LogSampleRunner runner = new LogSampleRunner(args);
    runner.uploadApp();
    runner.submitVMLogGeneratorApp();
    Thread.sleep(35000);
    runner.submitLogSampleDataflowChain();
    runner.submitVMLogValidatorApp().waitForTerminated(30000);;
  }
}

@@ -575,16 +575,62 @@
 IOUtil;%0A
+import com.neverwinterdp.vm.HadoopProperties;%0A
 import c
@@ -651,32 +651,32 @@
 dp.vm.VMConfig;%0A
-
 import com.never
@@ -804,16 +804,65 @@
 bmitter;
+%0Aimport com.neverwinterdp.vm.client.YarnVMClient;
 %0A%0Apublic
@@ -1153,24 +1153,29 @@
 .connect();%0A
+    %0A
     VMClient
@@ -1185,16 +1185,467 @@
 Client =
+ null;%0A    if(config.dfsAppHome != null) %7B%0A      String hadoopMaster = System.getProperty(%22shell.hadoop-master%22);%0A      HadoopProperties hadoopProps = new HadoopProperties() ;%0A      hadoopProps.put(%22yarn.resourcemanager.address%22, hadoopMaster + %22:8032%22);%0A      hadoopProps.put(%22fs.defaultFS%22, %22hdfs://%22 + hadoopMaster +%22:9000%22);%0A      vmClient = new YarnVMClient(registry, VMConfig.ClusterEnvironment.YARN, hadoopProps) ;%0A    %7D else %7B%0A      vmClient =
  new VMC
@@ -1661,16 +1661,22 @@
 istry);%0A
+    %7D%0A
     shel

fix vm client creation for real cluster