LoadingRunnable implements Runnable { public synchronized LoadStatistic getLoadStatistic() { final Long position = obserableInputStreamImpl.map(ObservableInputStreamImpl::getCurrentRead).orElse(0L); return new LoadStatistic(source, position, lastFileSize); } LoadingRunnable(Source source, LogImporter logImporter, LogDataCollector logDataCollector, long sleepTime, Optional<Long> bufferingTime, StatsService statsService);  LoadingRunnable(Source source, LogImporter logImporter, LogDataCollector logDataCollector, long sleepTime, Optional<Long> bufferingTime, Optional<AcceptCondition> withAcceptCondition, StatsService statsService); @Override void run(); synchronized LoadStatistic getLoadStatistic(); synchronized void pause(); synchronized void resume(); synchronized void stop(); synchronized void setFilter(Optional<AcceptCondition> acceptCondition);  }
@Test(invocationCount = 5) public void testLoadStatisticEmptyOnStartWithLogParser() { underTest = createLoadingRunnable(getJulLogParser()); Assert.assertEquals(underTest.getLoadStatistic().getPosition(), 0); executorService.submit(underTest); await().until(() -> underTest.getLoadStatistic().getPosition() == 0); await().until(() -> underTest.getLoadStatistic().getTotal() == 0); }
@Test(invocationCount = 5) public void testLoadStatisticAfterFullReadWithLogParser() throws Exception { underTest = createLoadingRunnable(getJulLogParser()); saveLines(Range.closed(0, julSimpleLogLines.size()), julSimpleLogLines, outputStream); executorService.submit(underTest); await().atMost(5, SECONDS).until(() -> underTest.getLoadStatistic().getPosition() == file.length()); await().atMost(5, SECONDS).until(() -> underTest.getLoadStatistic().getTotal() == file.length()); }