package com.github.dkanellis.skyspark.performance;

import com.github.dkanellis.skyspark.api.algorithms.SkylineAlgorithm;
import com.github.dkanellis.skyspark.api.helpers.SparkContextWrapper;
import com.github.dkanellis.skyspark.api.helpers.TextFileToPointRDD;
import com.github.dkanellis.skyspark.performance.parsing.Settings;
import com.github.dkanellis.skyspark.performance.result.PointDataFile;
import com.github.dkanellis.skyspark.performance.result.Result;
import com.github.dkanellis.skyspark.performance.result.ResultWriter;
import com.github.dkanellis.skyspark.performance.result.XmlResultWriter;
import com.google.common.base.Stopwatch;
import org.apache.spark.api.java.JavaRDD;

import java.awt.geom.Point2D;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class Main {

    private static Stopwatch stopwatch;
    private static TextFileToPointRDD textFileToPointRDD;
    private static Settings settings;
    private static ResultWriter resultWriter;

    public static void main(String[] args) {
        settings = Settings.fromArgs(args);
        if (settings == null) {
            return;
        }

        stopwatch = new Stopwatch();
        textFileToPointRDD = new TextFileToPointRDD(new SparkContextWrapper());
        resultWriter = new XmlResultWriter(settings.getOutputPath());

        settings.getAlgorithms().forEach(Main::executeAlgorithm);
    }

    private static void executeAlgorithm(SkylineAlgorithm skylineAlgorithm) {
        for (PointDataFile pointDataFile : settings.getPointDataFiles()) {
            executeWithFile(skylineAlgorithm, pointDataFile);
        }
    }

    private static void executeWithFile(SkylineAlgorithm skylineAlgorithm, PointDataFile pointDataFile) {
        for (int timesToRun = 0; timesToRun < settings.getTimes(); ++timesToRun) {
            JavaRDD<Point2D> points = textFileToPointRDD.getPointRddFromTextFile(pointDataFile.getFilePath(), " ");

            stopwatch.start();
            JavaRDD<Point2D> skylines = skylineAlgorithm.computeSkylinePoints(points);
            List<Point2D> skylineList = skylines.collect();
            stopwatch.stop();

            Result result = new Result(skylineAlgorithm.toString(), pointDataFile,
                    stopwatch.elapsed(TimeUnit.NANOSECONDS), skylineList.size());
            resultWriter.writeResult(result);

            stopwatch.reset();
        }
    }
}

@@ -773,16 +773,51 @@
 imeUnit;
+%0Aimport java.util.stream.IntStream;
 %0A%0Apublic
@@ -1279,16 +1279,34 @@
 Wrapper(
+%22test%22, %22local%5B8%5D%22
 ));%0A    
@@ -1384,54 +1384,96 @@
     
-settings.getAlgorithms().forEach(Main::execute
+IntStream.rangeClosed(1, settings.getTimes())%0A                .forEach(t -%3E settings.get
 Algo
@@ -1481,104 +1481,35 @@
 ithm
+s(
 )
-;
 %0A    
-%7D%0A%0A
     
-private static void executeAlgorithm(SkylineAlgorithm skylineAlgorithm) %7B%0A
+       
         
 for 
@@ -1508,42 +1508,25 @@
     
+ .
 for
- (PointDataFile pointDataFile :
+Each(algo -%3E
  set
@@ -1550,20 +1550,20 @@
 aFiles()
-) %7B%0A
+%0A   
         
@@ -1570,67 +1570,73 @@
     
-executeWithFile(skylineAlgorithm, pointDataFile);%0A        %7D
+                 .forEach(file -%3E executeWithFile(algo, file))));
 %0A   
@@ -1749,95 +1749,8 @@
 ) %7B%0A
-        for (int timesToRun = 0; timesToRun %3C settings.getTimes(); ++timesToRun) %7B%0A    
     
@@ -1858,36 +1858,32 @@
  %22 %22);%0A%0A        
-    
 stopwatch.start(
@@ -1889,28 +1889,24 @@
 ();%0A        
-  
-  
 JavaRDD%3CPoin
@@ -1964,28 +1964,24 @@
 ts(points);%0A
-    
         List
@@ -2024,36 +2024,32 @@
 lect();%0A        
-    
 stopwatch.stop()
@@ -2047,28 +2047,24 @@
 ch.stop();%0A%0A
-    
         Resu
@@ -2138,36 +2138,32 @@
                 
-    
 stopwatch.elapse
@@ -2177,12 +2177,13 @@
 nit.
-NANO
+MILLI
 SECO
@@ -2205,28 +2205,24 @@
 st.size());%0A
-    
         resu
@@ -2252,36 +2252,32 @@
 sult);%0A%0A        
- 
-   
 stopwatch.reset(
@@ -2279,26 +2279,16 @@
 eset();%0A
-        %7D%0A
     %7D%0A%7D%0A

Improve Main