package org.broadinstitute.hellbender.tools.spark.pipelines;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.broadinstitute.hellbender.cmdline.Argument;
import org.broadinstitute.hellbender.cmdline.CommandLineProgramProperties;
import org.broadinstitute.hellbender.cmdline.StandardArgumentDefinitions;
import org.broadinstitute.hellbender.cmdline.programgroups.SparkProgramGroup;
import org.broadinstitute.hellbender.engine.spark.GATKSparkTool;
import org.broadinstitute.hellbender.exceptions.UserException;
import org.broadinstitute.hellbender.tools.FlagStat.FlagStatus;
import org.broadinstitute.hellbender.utils.gcs.BucketUtils;
import org.broadinstitute.hellbender.utils.read.GATKRead;

import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintStream;

@CommandLineProgramProperties(summary ="runs FlagStat on Spark",
        oneLineSummary = "FlagStat on Spark",
        programGroup = SparkProgramGroup.class)
public final class FlagStatSpark extends GATKSparkTool {

    private static final long serialVersionUID = 1L;

    @Override
    public boolean requiresReads() { return true; }

    @Argument(doc = "uri for the output file: a local file path",
            shortName = StandardArgumentDefinitions.OUTPUT_SHORT_NAME, fullName = StandardArgumentDefinitions.OUTPUT_LONG_NAME,
            optional = true)
    public String out;

    @Override
    protected void runTool(final JavaSparkContext ctx) {
        final JavaRDD<GATKRead> reads = getReads();

        final FlagStatus result = reads.aggregate(new FlagStatus(), FlagStatus::add, FlagStatus::merge);
        System.out.println(result);
        final File file = new File(out);
        try(final OutputStream outputStream = BucketUtils.createNonGCSFile(file.getPath());
            final PrintStream ps = new PrintStream(outputStream)) {
            ps.print(result);
        } catch(final IOException e){
            throw new UserException.CouldNotCreateOutputFile(file, e);
        }
    }
}

@@ -1690,32 +1690,66 @@
 rintln(result);%0A
+%0A        if ( out != null ) %7B%0A    
         final Fi
@@ -1785,12 +1785,18 @@
     
-try(
+    try ( 
 fina
@@ -1879,24 +1879,30 @@
             
+     
+ 
 final PrintS
@@ -1941,20 +1941,25 @@
 tStream)
+ 
 ) %7B%0A
+    
         
@@ -1992,16 +1992,34 @@
     
-%7D
+    %7D%0A           
  catch
-(
+ ( 
 fina
@@ -2037,11 +2037,17 @@
 on e
-)%7B%0A
+ ) %7B%0A    
     
@@ -2109,24 +2109,38 @@
 e(file, e);%0A
+            %7D%0A
         %7D%0A  

Fix NPE in FlagStatSpark when -O is not specified