package au.gov.amsa.geo.adhoc;

import java.io.File;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;

import au.gov.amsa.risky.format.BinaryFixes;
import au.gov.amsa.risky.format.Fix;
import au.gov.amsa.util.Files;
import rx.Observable;

public class VesselsInGbrMain {

    public static void main(String[] args) {
        Pattern pattern = Pattern.compile(".*\\.track");
        List<File> list = new ArrayList<File>();
        list.addAll(Files.find(new File("/media/an/binary-fixes-5-minute/2014"), pattern));
        list.addAll(Files.find(new File("/media/an/binary-fixes-5-minute/2015"), pattern));
        list.addAll(Files.find(new File("/media/an/binary-fixes-5-minute/2016"), pattern));
        AtomicInteger count = new AtomicInteger();
        Observable.from(list) //
                .groupBy(f -> count.getAndIncrement() % Runtime.getRuntime().availableProcessors()) //
                .flatMap(files -> files.doOnNext(System.out::println) //
                        .flatMap(f -> BinaryFixes.from(f) //
                                .filter(fix -> inGbr(fix)) //
                                .map(fix -> fix.mmsi()) //
                                .distinct()))
                .distinct() //
                .toBlocking() //
                .forEach(System.out::println);
    }

    private static boolean inGbr(Fix fix) {
        return fix.lat() >= -27.8 && fix.lat() <= -8.4 && fix.lon() >= 142 && fix.lon() <= 162;
    }

}

@@ -40,20 +40,185 @@
 java.io.
-File
+BufferedWriter;%0Aimport java.io.File;%0Aimport java.io.FileNotFoundException;%0Aimport java.io.FileOutputStream;%0Aimport java.io.IOException;%0Aimport java.io.OutputStreamWriter
 ;%0Aimport
@@ -484,87 +484,397 @@
 le;%0A
-%0Apublic class VesselsInGbrMain %7B%0A%0A    public static void main(String%5B%5D args) %7B%0A
+import rx.Scheduler;%0Aimport rx.observables.GroupedObservable;%0Aimport rx.schedulers.Schedulers;%0A%0Apublic class VesselsInGbrMain %7B%0A%0A    public static void main(String%5B%5D args) throws IOException %7B%0A        File out = new File(%22target/mmsi.txt%22);%0A        out.delete();%0A        try (BufferedWriter b = new BufferedWriter(%0A                new OutputStreamWriter(new FileOutputStream(out)))) %7B%0A    
     
@@ -934,16 +934,20 @@
         
+    
 List%3CFil
@@ -975,24 +975,28 @@
 st%3CFile%3E();%0A
+    
         list
@@ -1075,32 +1075,36 @@
 tern));%0A        
+    
 list.addAll(File
@@ -1163,32 +1163,36 @@
 5%22), pattern));%0A
+    
         list.add
@@ -1275,16 +1275,20 @@
         
+    
 AtomicIn
@@ -1330,16 +1330,20 @@
         
+    
 Observab
@@ -1375,16 +1375,20 @@
         
+ 
+   
 .groupBy
@@ -1416,16 +1416,44 @@
 rement()
+%0A                           
  %25 Runti
@@ -1502,32 +1502,36 @@
                 
+    
 .flatMap(files -
@@ -1536,103 +1536,185 @@
  -%3E 
-files.doOnNext(System.out::println) //%0A                        .flatMap(f -%3E BinaryFixes.from(f
+vesselsInGbr(files, Schedulers.computation())).distinct() //%0A                    .sorted() //%0A                    .doOnNext(m -%3E write(b, m)) //%0A                    .toBlocking(
 ) //
@@ -1730,32 +1730,46 @@
             
+.subscribe();%0A
         
     .filter(
@@ -1760,41 +1760,77 @@
     
+%7D%0A
     
-.filter(fix -%3E inGbr(fix)) //
+%7D%0A%0A    private static void write(BufferedWriter b, Integer m) %7B
 %0A   
@@ -1826,33 +1826,38 @@
 er m) %7B%0A        
- 
+try %7B%0A
                 
@@ -1856,45 +1856,139 @@
     
-           .map(fix -%3E fix.mmsi()) //
+b.write(String.valueOf(m));%0A            b.write('%5Cn');%0A        %7D catch (IOException e) %7B%0A            throw new RuntimeException(e);
 %0A   
@@ -1996,105 +1996,321 @@
     
+%7D%0A
     
+%7D%0A%0A
     
-                .distinct()))%0A                .distinct() //%0A                .toBlocking(
+private static Observable%3CInteger%3E vesselsInGbr(GroupedObservable%3CInteger, File%3E files,%0A            Scheduler scheduler) %7B%0A        return files //%0A                .flatMap(f -%3E BinaryFixes.from(f) //%0A                        .filter(fix -%3E inGbr(fix)) //%0A                        .map(fix -%3E fix.mmsi()
 ) //
@@ -2330,36 +2330,50 @@
     
+    
-.forEach(System.out::println
+    .distinct().subscribeOn(scheduler)
 );%0A 

add VesselsInGbrMain