package au.gov.amsa.risky.format;

import java.io.File;
import java.util.List;
import java.util.regex.Pattern;

import rx.Observable;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;
import au.gov.amsa.risky.format.BinaryFixes;
import au.gov.amsa.util.Files;

import com.github.davidmoten.rx.slf4j.Logging;

public class BinaryFixesMain {

	public static void main(String[] args) {
		//perform a speed test for loading BinaryFixes from disk
		
		List<File> files = Files.find(new File(
				"/media/an/binary-fixes/2014-year-downsample-5-mins"), Pattern
				.compile(".*\\.track"));
		long t = System.currentTimeMillis();
		long count = Observable
		// list files
				.from(files)
				// count each file asynchronously
				.flatMap(new Func1<File, Observable<Long>>() {
					@Override
					public Observable<Long> call(File file) {
						return BinaryFixes.from(file).countLong()
								.subscribeOn(Schedulers.computation());
					}
				})
				// total counts
				.scan(0L, new Func2<Long, Long, Long>() {
					@Override
					public Long call(Long a, Long b) {
						return a + b;
					}
				})
				// log count so far
				.lift(Logging.<Long> logger().showCount().prefix("records=")
						.showMemory().every(1000).log())
				// get last count (total)
				.last()
				//block and get
				.toBlocking().single();
		long elapsed = System.currentTimeMillis() - t;
		System.out.println("Total records = " + count + ", numPerSecond="
				+ count * 1000.0 / elapsed + ", timeMs=" + elapsed);
	}
}

@@ -70,14 +70,119 @@
 til.
+Array
 List;%0A
+import java.util.Collections;%0Aimport java.util.List;%0Aimport java.util.concurrent.ConcurrentHashMap;%0A
 impo
@@ -228,24 +228,53 @@
 Observable;%0A
+import rx.functions.Action1;%0A
 import rx.fu
@@ -352,53 +352,8 @@
 rs;%0A
-import au.gov.amsa.risky.format.BinaryFixes;%0A
 impo
@@ -564,16 +564,110 @@
 disk%0A%09%09%0A
+%09%09final ConcurrentHashMap%3CLong, List%3CFix%3E%3E map = new ConcurrentHashMap%3CLong, List%3CFix%3E%3E();%0A%09%09%0A
 %09%09List%3CF
@@ -1080,16 +1080,97 @@
 om(file)
+%0A%09%09%09%09%09%09%09%09//add to map%0A%09%09%09%09%09%09%09%09//.doOnNext(addToMap(map))%0A%09%09%09%09%09%09%09%09//count%0A%09%09%09%09%09%09%09%09
 .countLo
@@ -1221,32 +1221,38 @@
 tion());%0A%09%09%09%09%09%7D%0A
+%09%09%09%09%09%0A
 %09%09%09%09%7D)%0A%09%09%09%09// to
@@ -1664,16 +1664,66 @@
 () - t;%0A
+%09%09System.out.println(%22Map size = %22 + map.size());%0A
 %09%09System
@@ -1782,16 +1782,16 @@
 econd=%22%0A
-
 %09%09%09%09+ co
@@ -1842,10 +1842,413 @@
 ed);%0A%09%7D%0A
+%09%0A%09private static Action1%3C? super Fix%3E addToMap(final ConcurrentHashMap%3CLong, List%3CFix%3E%3E map)%7B%0A%09%09return new Action1%3CFix%3E() %7B%0A%09%09%09@Override%0A%09%09%09public void call(Fix fix) %7B%0A%09%09%09%09%09List%3CFix%3E list = map.get(fix.getMmsi());%0A%09%09%09%09%09if (list==null) %7B%0A%09%09%09%09%09%09map.putIfAbsent(fix.getMmsi(), Collections.synchronizedList( new ArrayList%3CFix%3E()));%0A%09%09%09%09%09%09list = map.get(fix.getMmsi());%0A%09%09%09%09%09%7D%0A%09%09%09%09%09list.add(fix);%0A%09%09%09%7D%7D;%0A%09%7D%0A
 %7D%0A

load binary fixes dev