package au.gov.amsa.ais.rx;

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

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import rx.Observable;
import rx.Observer;
import rx.functions.Func1;
import rx.schedulers.Schedulers;
import au.gov.amsa.ais.AisMessage;
import au.gov.amsa.ais.Timestamped;

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

public class BinaryFixesWriterMain {

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

	public static void main(String[] args) throws IOException {
		log.info("starting");
		final String inputFilename;
		if (args.length > 0)
			inputFilename = args[0];
		else
			inputFilename = "G:\\mariweb";
		final String outputFilename;
		if (args.length >= 2)
			outputFilename = args[1];
		else
			outputFilename = "target/binary";
		final String pattern;
		if (args.length >= 3)
			pattern = args[2];
		else
			pattern = "NMEA_ITU_(201412|20150101).*.gz";
		// pattern = "NMEA.*.gz";

		log.info("Converting NMEA files in " + inputFilename);
		log.info("to BinaryFixes format in " + outputFilename);
		log.info("using pattern='" + pattern + "'");

		File input = new File(inputFilename);
		File output = new File(outputFilename);
		long t = System.currentTimeMillis();

		int logEvery = 100000;
		int writeBufferSize = 1000;
		Pattern inputPattern = Pattern.compile(pattern);

		if (true) {

			BinaryFixesWriter
					.writeFixes(input, inputPattern, output, logEvery,
							writeBufferSize).observeOn(Schedulers.immediate())
					.subscribe(new Observer<Integer>() {

						@Override
						public void onCompleted() {

						}

						@Override
						public void onError(Throwable e) {
							e.printStackTrace();
						}

						@Override
						public void onNext(Integer t) {
							// TODO Auto-generated method stub

						}
					});
		} else {
			// read 11 million NMEA lines
			Streams.nmeaFromGzip("/media/analysis/test/NMEA_ITU_20150101.gz")
			// buffer in groups of 20,000 to assign to computation
			// threads
					.buffer(20000)
					// parse the messages asynchronously using computation
					// scheduler
					.flatMap(
							new Func1<List<String>, Observable<Timestamped<AisMessage>>>() {
								@Override
								public Observable<Timestamped<AisMessage>> call(
										List<String> list) {
									return Streams.extractMessages(
											Observable.from(list))
									// do async
											.subscribeOn(
													Schedulers.computation());
								}
							})
					// log stuff
					.lift(Logging.<Timestamped<AisMessage>> logger()
							.showRateSince("rate=", 1000).showCount()
							.every(100000).log())
					// count emitted
					.count()
					// observer results in current thread
					.observeOn(Schedulers.immediate())
					// go
					.subscribe(new Observer<Integer>() {

						@Override
						public void onCompleted() {

						}

						@Override
						public void onError(Throwable e) {
							e.printStackTrace();
							throw new RuntimeException(e);
						}

						@Override
						public void onNext(Integer n) {
							System.out.println(n + " items");
						}
					});
		}

		log.info("finished in " + (System.currentTimeMillis() - t) / 1000.0
				+ "s");
	}
}

@@ -656,86 +656,24 @@
 name
-;%0A%09%09if (args.length %3E 0)%0A%09%09%09inputFilename = args%5B0%5D;%0A%09%09else%0A%09%09%09inputFilename =
+ = prop(%22input%22,
  %22G:
@@ -682,16 +682,17 @@
 mariweb%22
+)
 ;%0A%09%09fina
@@ -718,89 +718,25 @@
 name
-;%0A%09%09if (args.length %3E= 2)%0A%09%09%09outputFilename = args%5B1%5D;%0A%09%09else%0A%09%09%09outputFilename =
+ = prop(%22output%22,
  %22ta
@@ -747,16 +747,17 @@
 /binary%22
+)
 ;%0A%09%09fina
@@ -776,75 +776,26 @@
 tern
-;%0A%09%09if (args.length %3E= 3)%0A%09%09%09pattern = args%5B2%5D;%0A%09%09else%0A%09%09%09
+ = prop(%22
 pattern
- =
+%22,
  %22NM
@@ -805,59 +805,19 @@
 ITU_
-(
 2014
-12%7C20150101).*.gz%22;%0A%09%09// pattern = %22NMEA
 .*.gz%22
+)
 ;%0A%0A%09
@@ -3041,11 +3041,128 @@
 %22s%22);%0A%09%7D
+%0A%0A%09private static String prop(String name, String defaultValue) %7B%0A%09%09return System.getProperty(name, defaultValue);%0A%09%7D
 %0A%7D%0A

fix pom.xml