package au.gov.amsa.geo.adhoc;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.io.UncheckedIOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.TimeZone;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
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 au.gov.amsa.risky.format.Fix;
import rx.Observable;

public final class BinaryFixesWithMmsiGzCombinedSortMain {
    
    private static final Logger log = Logger.getLogger(BinaryFixesWithMmsiGzCombinedSortMain.class);

    public static void main(String[] args) throws InterruptedException {
        File tracks = new File("/home/dxm/AIS/tracks");
        File combinedSortedTracks = new File("/home/dxm/combinedSortedTracks");
        combinedSortedTracks.mkdir();

        AtomicReference<FileFixes> previous = new AtomicReference<>();
        AtomicInteger n = new AtomicInteger();
        Observable //
                .from(tracks.listFiles()) //
                .filter(x -> x.getName().endsWith(".track.gz") && x.getName().compareTo("2019-08-06") >= 0) //
                .map(x -> new FileFixes(x,
                        BinaryFixes.from(x, true, BinaryFixesFormat.WITH_MMSI).toList().toBlocking().first()))
                .doOnNext(ff -> {
                    FileFixes prev = previous.get();
                    if (prev != null) {
                        long removed = 0;
                        long added = 0;
                        Iterator<Fix> it = ff.fixes.iterator();
                        while (it.hasNext()) {
                            Fix fix = it.next();
                            if (fix.time() < ff.startTime) {
                                it.remove();
                                removed++;
                                if (fix.time() >= prev.startTime) {
                                    prev.fixes.add(fix);
                                    added++;
                                }
                            }
                        }
                        prev.fixes.sort((x, y) -> Long.compare(x.time(), y.time()));
                        File f = new File(combinedSortedTracks, prev.file.getName());
                        try (OutputStream out = new GZIPOutputStream(new FileOutputStream(f))) {
                            for (Fix fix : prev.fixes) {
                                BinaryFixes.write(fix, out, BinaryFixesFormat.WITH_MMSI);
                            }
                        } catch (IOException e) {
                            throw new UncheckedIOException(e);
                        }
                        log.info(n.incrementAndGet() + ": removed " + removed + ", added " + added + " to " + f);
                    }
                    previous.set(ff);
                }).subscribe();
    }

    private static final class FileFixes {
        final File file;
        final List<Fix> fixes;
        final long startTime;

        FileFixes(File file, List<Fix> fixes) {
            this.file = file;
            this.fixes = fixes;
            String datePart = file.getName().substring(0, 10);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-mm-dd");
            sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
            try {
                startTime = sdf.parse(datePart).getTime();
            } catch (ParseException e) {
                throw new RuntimeException(e);
            }
        }
    }

}

@@ -262,17 +262,16 @@
 til.
-ArrayList
+Iterator
 ;%0Aim
@@ -281,32 +281,34 @@
 t java.util.
-Iterator
+LinkedList
 ;%0Aimport jav
@@ -453,44 +453,8 @@
 ce;%0A
-import java.util.stream.Collectors;%0A
 impo
@@ -1045,24 +1045,25 @@
 SortedTracks
+2
 %22);%0A        
@@ -1360,25 +1360,26 @@
 e().
-compareTo
+startsWith
 (%222019-0
 8-06
@@ -1378,19 +1378,13 @@
 19-0
-8
+1
 -0
-6
 %22)
- %3E= 0
 ) //
@@ -1405,134 +1405,522 @@
    .
-map(x -%3E new FileFixes(x,%0A                        BinaryFixes.from(x, true, BinaryFixesFormat.WITH_MMSI).toList().toBlocking()
+sorted((x,y) -%3E x.getName().compareTo(y.getName())) //%0A                .map(x -%3E new FileFixes(x, BinaryFixes.from(x, true, BinaryFixesFormat.WITH_MMSI) //%0A                        // use a linked list because iterator.remove is efficient with that data structure%0A                        .reduce(new LinkedList%3CFix%3E(), (list, fix) -%3E %7B%0A                            list.add(fix);%0A                            return list;%0A                        %7D) //%0A                        .toBlocking() //%0A                        
 .fir
@@ -2817,25 +2817,219 @@
     
-File f = new File
+writeFixes(combinedSortedTracks, n, prev, removed, added);%0A                    %7D%0A                    previous.set(ff);%0A                %7D).subscribe();%0A        if (previous.get() != null) %7B%0A            writeFixes
 (com
@@ -3050,28 +3050,32 @@
 cks,
+ n,
  prev
-.file.getName()
+ious.get(), 0, 0
 );%0A 
@@ -3077,32 +3077,238 @@
 0);%0A        
+%7D%0A
     
+%7D%0A%0A
     
+private static void writeFixes(File combinedSortedTracks, AtomicInteger n, FileFixes prev, long removed,%0A            long added) %7B%0A        File f = new File(combinedSortedTracks, prev.file.getName());%0A
         try 
@@ -3376,32 +3376,16 @@
 (f))) %7B%0A
-                
         
@@ -3433,32 +3433,16 @@
         
-         
-       
 BinaryFi
@@ -3507,42 +3507,10 @@
     
-                %7D%0A                
+%7D%0A
     
@@ -3539,32 +3539,16 @@
 on e) %7B%0A
-                
         
@@ -3598,42 +3598,10 @@
     
-                %7D%0A                
+%7D%0A
     
@@ -3646,16 +3646,30 @@
 removed 
+from next day=
 %22 + remo
@@ -3683,16 +3683,28 @@
 , added 
+to this day=
 %22 + adde
@@ -3724,100 +3724,8 @@
 f);%0A
-                    %7D%0A                    previous.set(ff);%0A                %7D).subscribe();%0A
     

use linkedlist to increase perf of removal