### Question:
DistanceTravelledCalculator { public Observable<CellAndDistance> calculateDistanceByCellFromFiles(Observable<File> files) { int numFiles = files.count().toBlocking().single(); log.info("numFiles=" + numFiles); AtomicLong fileCount = new AtomicLong(); AtomicLong cellCount = new AtomicLong(1); return files .buffer(Math.max(1, (int) Math.round( Math.ceil(numFiles / Runtime.getRuntime().availableProcessors())))) .flatMap(fileList -> extractCellDistances(fileCount, cellCount, fileList)) .collect(bigMapFactory(), collectCellDistances()) .flatMap(listCellDistances()) .doOnNext(sumNauticalMiles()); } DistanceTravelledCalculator(Options options, DistanceCalculationMetrics metrics); Observable<CellAndDistance> calculateDistanceByCellFromFiles(Observable<File> files); DistanceCalculationMetrics getMetrics(); static Observable<CellValue> calculateDensityByCellFromFiles(Options options,
            Observable<File> files, int horizontal, int vertical,
            DistanceCalculationMetrics metrics); static CalculationResult calculateTrafficDensity(Options options,
            Observable<File> files); static CalculationResult calculateTrafficDensity(Options options, Observable<File> files,
            int horizontal, int vertical); static void saveCalculationResultAsText(Options options,
            CalculationResult calculationResult, String filename); static void saveCalculationResultAsNetcdf(Options options,
            CalculationResult calculationResult, String filename); static Observable<Options> partition(final Options options, final int horizontal,
            final int vertical);  }### Answer:
@Test public void testWithFiles() { long t = System.currentTimeMillis(); Observable<File> files = from( new File("src/test/resources/fixes").listFiles(new FileFilter() { @Override public boolean accept(File file) { return file.getName().startsWith("503.*track"); } })); final AtomicInteger count = new AtomicInteger(); DistanceCalculationMetrics metrics = new DistanceCalculationMetrics(); new DistanceTravelledCalculator(Options.builder().originLat(0).originLon(0) .cellSizeDegrees(0.1).bounds(new Bounds(0, 120, -60, 175)).build(), metrics) .calculateDistanceByCellFromFiles(files).toBlocking() .forEach(new Action1<CellAndDistance>() { @Override public void call(CellAndDistance cell) { System.out.println(cell); count.incrementAndGet(); } }); System.out.println( count.get() + " cells returned in " + (System.currentTimeMillis() - t) + "ms"); }