package diergo.csv;

import com.tngtech.java.junit.dataprovider.DataProvider;
import com.tngtech.java.junit.dataprovider.DataProviderRunner;
import org.junit.Test;
import org.junit.runner.RunWith;

import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadMXBean;
import java.util.Collection;
import java.util.List;
import java.util.stream.LongStream;
import java.util.stream.Stream;

import static diergo.csv.CsvParserBuilder.csvParser;
import static diergo.csv.ErrorHandlers.ignoreErrors;
import static diergo.csv.Readers.asLines;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.stream.IntStream.rangeClosed;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.lessThan;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;

@RunWith(DataProviderRunner.class)
public class CsvReaderPerformanceTest {

    public static final String MAXMIND_WORLD_CITIES_POP = "/worldcitiespop.txt";

    @Test
    @DataProvider({"true","false"})
    public void readMillions(boolean usingFlatMap) throws IOException {
        String kind = usingFlatMap ? "using flat map" : "using filter and map";
        System.out.println("starting dry run " + kind + "…");
        runOnce(usingFlatMap);
        long[] times = new long[6];
        rangeClosed(1, times.length).forEachOrdered(loop -> {
            try {
                System.out.print(String.format("loop %d", loop));
                long[] timeAndCount = runOnce(usingFlatMap);
                times[loop - 1] = timeAndCount[0];
                System.out.println(String.format(" took %dms to read %d rows", timeAndCount[0], timeAndCount[1]));
                assertThat(timeAndCount[1], greaterThan(3000000L));
            } catch (IOException e) {
                fail(e.getMessage());
            }
        });
        double average = LongStream.of(times).average().getAsDouble();
        System.out.println(String.format("average %s is %.0fms", kind, average));
        assertThat("acceptable average[ms]", average, lessThan(2000.0));
    }
    
    private long[] runOnce(boolean usingFlatMap) throws IOException {
        InputStreamReader worldCitiesPopulation = new InputStreamReader(getClass().getResourceAsStream(MAXMIND_WORLD_CITIES_POP), UTF_8);
        ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean();
        long start = threadMXBean.getCurrentThreadCpuTime();
        Stream<List<Row>> intermediate = asLines(worldCitiesPopulation)
                .map(csvParser().separatedBy(',').handlingErrors(ignoreErrors()).build());
        long count = (usingFlatMap ? intermediate.flatMap(Collection::stream) :
                intermediate.filter(rows -> !rows.isEmpty()).map(rows -> rows.get(0)))
                .count();
        long durations = (threadMXBean.getCurrentThreadCpuTime() - start) / 1000000L;
        try {
            return new long[]{durations, count};
        } finally {
            worldCitiesPopulation.close();
        }
    }
}

@@ -1092,16 +1092,69 @@
 true
-%22,%22false
+,false,2500%22,%22false,false,2500%22,%22true,true,50%22,%22false,true,50
 %22%7D)%0A
@@ -1194,32 +1194,64 @@
 ean usingFlatMap
+, boolean parallel, long maxTime
 ) throws IOExcep
@@ -1279,16 +1279,17 @@
  kind = 
+(
 usingFla
@@ -1336,16 +1336,77 @@
 and map%22
+)%0A            + %22, %22 + (parallel ? %22parallel%22 : %22sequential%22)
 ;%0A      
@@ -1481,32 +1481,42 @@
 nce(usingFlatMap
+, parallel
 );%0A        long%5B
@@ -1742,16 +1742,26 @@
 gFlatMap
+, parallel
 );%0A     
@@ -2311,14 +2311,23 @@
 han(
-2000.0
+(double)maxTime
 ));%0A
@@ -2384,16 +2384,34 @@
 gFlatMap
+, boolean parallel
 ) throws
@@ -2716,31 +2716,21 @@
 eam%3C
-List%3CRow%3E
+String
 %3E 
+l
 in
-termediat
 e
+s
  = a
@@ -2758,16 +2758,136 @@
 ulation)
+;%0A        if (parallel) %7B%0A            lines = lines.parallel();%0A        %7D%0A        Stream%3CList%3CRow%3E%3E intermediate = lines
 %0A       
@@ -3137,20 +3137,16 @@
 et(0)))%0A
-    
         

test also parallel