package au.gov.amsa.geo.adhoc;

import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.util.zip.GZIPOutputStream;

import org.apache.log4j.Logger;

import au.gov.amsa.risky.format.BinaryFixes;
import au.gov.amsa.risky.format.BinaryFixesFormat;
import rx.Observable;

public class BinaryFixesWithMmsiGzSorterMain {

    private static final Logger log = Logger.getLogger(BinaryFixesWithMmsiGzSorterMain.class);

    public static void main(String[] args) {

        File tracksDir = new File("/home/dxm/AIS/tracks");
        File sortedTracks = new File("/home/dxm/AIS/tracks-sorted");

        BinaryFixes.from(new File(tracksDir, "2019-01-01.track.gz"), true, BinaryFixesFormat.WITH_MMSI) //
                .doOnNext(System.out::println) //
                .subscribe();

        sortedTracks.mkdir();
        Observable<File> files = Observable //
                .from(tracksDir.listFiles()) //
                .filter(x -> x.getName().endsWith(".track.gz"));
        files.flatMap(x -> Observable //
                .just(x) //
                .doOnNext(f -> log.info("sorting " + f)) //
                .doOnNext(f -> sort(f, sortedTracks))) //
                .doOnNext(f -> log.info("sorted " + f)) //
                .toBlocking() //
                .subscribe();
    }

    private static void sort(File file, File outputDir) {
        File sorted = new File(outputDir, file.getName());
        try (OutputStream out = new GZIPOutputStream(new BufferedOutputStream(new FileOutputStream(sorted)))) {
            BinaryFixes.from(file, true, BinaryFixesFormat.WITH_MMSI) //
                    .doOnNext(System.out::println) //
                    .toSortedList((x, y) -> Long.compare(x.time(), y.time())) //
                    .flatMap(x -> Observable.from(x)) //
                    .doOnNext(fix -> BinaryFixes.write(fix, out, BinaryFixesFormat.WITH_MMSI)) //
                    .toBlocking() //
                    .subscribe();
        } catch (IOException e) {
            throw new UncheckedIOException(e);
        }
    }

}

@@ -202,32 +202,82 @@
 kedIOException;%0A
+import java.util.concurrent.atomic.AtomicInteger;%0A
 import java.util
@@ -450,16 +450,49 @@
 ervable;
+%0Aimport rx.schedulers.Schedulers;
 %0A%0Apublic
@@ -808,196 +808,8 @@
 );%0A%0A
-        BinaryFixes.from(new File(tracksDir, %222019-01-01.track.gz%22), true, BinaryFixesFormat.WITH_MMSI) //%0A                .doOnNext(System.out::println) //%0A                .subscribe();%0A%0A
     
@@ -990,24 +990,71 @@
 rack.gz%22));%0A
+        AtomicInteger n = new AtomicInteger();%0A
         file
@@ -1131,32 +1131,112 @@
    .
-doOnNext(f -%3E log.info(%22
+observeOn(Schedulers.computation()) //%0A                .doOnNext(f -%3E log.info(n.incrementAndGet() + %22: 
 sort
@@ -1240,24 +1240,34 @@
 orting %22 + f
+.getName()
 )) //%0A      
@@ -1358,17 +1358,29 @@
 og.info(
-%22
+n.get() + %22: 
 sorted %22
@@ -1383,16 +1383,26 @@
 ed %22 + f
+.getName()
 )) //%0A  
@@ -1775,62 +1775,8 @@
  //%0A
-                    .doOnNext(System.out::println) //%0A
     

add logging