TopologyRepository.java

@Service
@Slf4j
@RequiredArgsConstructor
public class TopologyRepository {

    private final CacheMono<TopologyRef, TopologyDto, TopologyDto> cache;
    private final TopologyLoader topologyLoader;
    private final TopologyCreator topologyCreator;

    @Autowired
    public UnoTopologyRepository(TopologyLoader topologyLoader,
                                 TopologyCreator topologyCreator) {
        this.topologyLoader = topologyLoader;
        this.topologyCreator = topologyCreator;
        cache = CacheMono.fromSupplier(this::retrieveTopology);
    }

    /**
     * Finds a topology from this repository by reference.
     */
    public Mono<TopologyDto> findUnoTopology(TopologyRef topologyRef) {
        return cache.lookup(topologyRef)
                .doOnNext(topology ->
                          log.info("Topology was found by lookup with key {}", topologyRef))
                .onErrorResume(err -> {
                    log.error("Error on lookup Topology by key {}, message: {}",
                              topologyRef, err.getMessage());
                    return Mono.empty();
                });
    }

    private Mono<TopologyDto> retrieveTopology(TopologyRef topologyRef) {
        CompletableFuture<UnoTopologyDto> future = CompletableFuture.supplyAsync(() -> {
            final var loaderContext = topologyLoader.retrieveTopology(topologyRef);
            return topologyCreator.createTopology(loaderContext);
        });
        return Mono.fromFuture(future);
    }
}