class 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 public  DefaultLocalMasterInstanceResolver(LocalMasterReadinessResolver localMasterReadinessResolver,
                                              MasterInstance initial);

}