AsyncDataWriter implements DataWriter<T> { @Override public Future<T> loadAndGet() { String correlationId = "" + System.currentTimeMillis() + ":" + r.nextLong(); Supplier<MapX<String, String>> dataMap = () -> MapX.fromMap(HashMapBuilder.map(MANIFEST_COMPARATOR_DATA_LOADER_KEY, comparator.toString()) .build()); return Future.of(() -> Tuple.tuple(comparator.load(), comparator.getData()), executorService) .peek(t -> bus.post(SystemData.<String, String> builder() .correlationId(correlationId) .dataMap(dataMap.get()) .errors(0) .processed(t._1() ? 1 : 0) .build())) .map(t -> t._2()); }  static AsyncDataWriter<T> asyncDataWriter(Executor executorService, ManifestComparator<T> comparator); static AsyncDataWriter<T> asyncDataWriter(Executor executorService, ManifestComparator<T> comparator,
            EventBus bus); @Override Future<T> loadAndGet(); @Override Future<Void> saveAndIncrement(T data); @Override Future<Boolean> isOutOfDate(); static final String MANIFEST_COMPARATOR_DATA_LOADER_KEY; static final String MANIFEST_COMPARATOR_DATA_WRITER_KEY; }
@Test public void testLoadAndGet() { assertThat(eventRecieved.get(), equalTo(0)); dummyMc.setData("hello world"); Future<String> res = writer.loadAndGet(); assertThat(res.get(), equalTo(Try.success("hello world"))); assertThat(dummyMc.loadCalled.get(), equalTo(1)); assertThat(eventRecieved.get(), equalTo(1)); }