DefaultLocalMasterInstanceResolver implements LocalMasterInstanceResolver { @Override public Flux<MasterInstance> observeLocalMasterInstanceUpdates() { return Flux.defer(() -> { AtomicReference<MasterInstance> lastRef = new AtomicReference<>(initial); return localMasterReadinessResolver.observeLocalMasterReadinessUpdates() .flatMap(update -> { Optional<MasterInstance> refreshedOpt = refresh(lastRef.get(), update); if (refreshedOpt.isPresent()) { MasterInstance refreshed = refreshedOpt.get(); lastRef.set(refreshed); return Flux.just(refreshed); } return Flux.empty(); }); }); } @Inject  DefaultLocalMasterInstanceResolver(LocalMasterReadinessResolver localMasterReadinessResolver,
                                              MasterInstance initial); @Override Flux<MasterInstance> observeLocalMasterInstanceUpdates();  }
@Test(timeout = 30_000) public void testResolver() { Iterator<MasterInstance> it = resolver.observeLocalMasterInstanceUpdates().toIterable().iterator(); readinessUpdatesProcessor.onNext(newReadinessStatus(ReadinessState.NotReady)); assertThat(it.next().getStatus().getState()).isEqualTo(MasterState.Starting); readinessUpdatesProcessor.onNext(newReadinessStatus(ReadinessState.Enabled)); assertThat(it.next().getStatus().getState()).isEqualTo(MasterState.NonLeader); }