package com.neverwinterdp.scribengin.dataflow;


import static org.junit.Assert.assertEquals;

import java.util.List;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import com.neverwinterdp.registry.Registry;
import com.neverwinterdp.scribengin.ScribenginClient;
import com.neverwinterdp.scribengin.builder.ScribenginClusterBuilder;
import com.neverwinterdp.scribengin.client.shell.ScribenginShell;
import com.neverwinterdp.scribengin.dataflow.test.KafkaDataflowTest;
import com.neverwinterdp.scribengin.service.ScribenginService;
import com.neverwinterdp.scribengin.tool.EmbededVMClusterBuilder;
import com.neverwinterdp.yara.snapshot.MetricRegistrySnapshot;

public class DataflowKafkaToKafkaUnitTest {
  static {
    System.setProperty("java.net.preferIPv4Stack", "true") ;
    System.setProperty("log4j.configuration", "file:src/test/resources/test-log4j.properties") ;
  }
  
  protected ScribenginClusterBuilder clusterBuilder ;
  protected ScribenginShell shell;
  
  @Before
  public void setup() throws Exception {
    clusterBuilder = new ScribenginClusterBuilder(new EmbededVMClusterBuilder()) ;
    clusterBuilder.clean(); 
    clusterBuilder.startVMMasters();
    clusterBuilder.startScribenginMasters();
    shell = new ScribenginShell(clusterBuilder.getVMClusterBuilder().getVMClient());
  }
  
  @After
  public void teardown() throws Exception {
    clusterBuilder.shutdown();
  }
  
  @Test
  public void testDataflows() throws Exception {
    DataflowSubmitter submitter = new DataflowSubmitter();
    submitter.start();

    Thread.sleep(5000); //make sure that the dataflow start and running;

    try {
      ScribenginClient scribenginClient = shell.getScribenginClient();
      assertEquals(2, scribenginClient.getScribenginMasters().size());
      submitter.waitForTermination(120000);
      Registry registry = shell.getVMClient().getRegistry();
      String dataflowPath = ScribenginService.getDataflowPath("kafka-to-kafka-1");
      List<MetricRegistrySnapshot> snapshots = DataflowRegistry.getMetrics(registry, dataflowPath) ;
      System.out.println(MetricRegistrySnapshot.getFormattedText(snapshots));
    } catch(Throwable err) {
      throw err;
    } finally {
      if(submitter.isAlive()) submitter.interrupt();
      Thread.sleep(3000);
    }
  }
  
  public class DataflowSubmitter extends Thread {
    public void run() {
      try {
        String command =
            "dataflow-test " + KafkaDataflowTest.TEST_NAME +
            " --dataflow-id    kafka-to-kafka-1" +
            " --dataflow-name  kafka-to-kafka" +
            " --worker 3" +
            " --executor-per-worker 1" +
            " --duration 120000" +
            " --task-max-execute-time 10000" +
            " --source-name input" +
            " --source-num-of-stream 10" +
            " --source-write-period 0" +
            " --source-max-records-per-stream 1000" +
            " --sink-name output " +
            " --print-dataflow-info -1" +
            " --debug-dataflow-task " +
            " --debug-dataflow-vm " +
            " --debug-dataflow-activity " +
            " --junit-report build/junit-report.xml" + 
            " --dump-registry";
        shell.execute(command);
      } catch(Exception ex) {
        ex.printStackTrace();
      } finally {
        notifyTermimation() ;
      }
    }
    
    synchronized void notifyTermimation() {
      notify();
    }
    
    synchronized void waitForTermination(long timeout) throws InterruptedException {
      wait(timeout);
    }
  }
}
@@ -1835,331 +1835,8 @@
 0);%0A
-      Registry registry = shell.getVMClient().getRegistry();%0A      String dataflowPath = ScribenginService.getDataflowPath(%22kafka-to-kafka-1%22);%0A      List%3CMetricRegistrySnapshot%3E snapshots = DataflowRegistry.getMetrics(registry, dataflowPath) ;%0A      System.out.println(MetricRegistrySnapshot.getFormattedText(snapshots));%0A
     

update report