package au.gov.amsa.risky.format;

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Observable.Transformer;
import rx.functions.Func2;

public class Downsample implements Transformer<Fix, Fix> {

	private long maxTimeBetweenFixesMs;

	public Downsample(long minTimeBetweenFixesMs) {
		this.maxTimeBetweenFixesMs = minTimeBetweenFixesMs;
	}

	public static Downsample minTimeStep(long duration, TimeUnit unit) {
		return new Downsample(unit.toMillis(duration));
	}

	@Override
	public Observable<Fix> call(Observable<Fix> fixes) {
		return fixes.scan(new Func2<Fix, Fix, Fix>() {
			@Override
			public Fix call(Fix latest, Fix fix) {
				if (fix.getTime() - latest.getTime() > maxTimeBetweenFixesMs)
					return fix;
				else
					return latest;
			}
		})
		// throw away repeats
				.distinct();
	}
}

@@ -670,16 +670,256 @@
 etTime()
+ %3C latest.getTime())%0A%09%09%09%09%09throw new RuntimeException(%22not in ascending time order!%22);%0A%09%09%09%09else if (fix.getMmsi() != latest.getMmsi())%0A%09%09%09%09%09throw new RuntimeException(%0A%09%09%09%09%09%09%09%22can only downsample a single vessel%22);%0A%09%09%09%09else if (fix.getTime()
  - lates

add tests for time order and mmsi to downsample