ova2

@Service
@Slf4j
@RequiredArgsConstructor
public class SspDefinitionenStore implements SspDefinitionConsumer {

    private CacheMono<VersionedId, SspDefinition, SspDefinition> sspDefinitionCache;
    private FluxSink<SspDefinition> sspDefinitionSink;

    @PostConstruct
    public void initialize() {

ova2

@Service
@Slf4j
@RequiredArgsConstructor
public class TopologyRepository {

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

    @Autowired

ova2

/**
 * Gets cached values as Java Stream. Returned stream is not sorted.
 */
public Stream<OVALUE> getValues() {
    final Lock readLock = lock.readLock();
    readLock.lock();
    try {
        return cache.values().stream().flatMap(cachedValue -> cachedValue.getValue().stream());
    } finally {
        readLock.unlock();

ova2

/**
 * Finds a value by key in an in-memory cache or load it from a remote source.
 * The loaded value will be cached.
 */
public Mono<OVALUE> lookup(KEY key) {
    return Mono.defer(() -> getValueAsMono(key)
            .switchIfEmpty(Mono.defer(() -> onCacheMissResume(key)))
    );
}

ova2

private static class CacheMonoValue<VALUE> {

    private Mono<VALUE> mono;
    private Signal<VALUE> signal;

    CacheMonoValue(Mono<VALUE> mono) {
        this.mono = mono;
    }

    CacheMonoValue(Signal<VALUE> signal) {

ova2

@Slf4j
public class CacheMono<KEY, IVALUE, OVALUE> {

    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    private final Map<KEY, CacheMonoValue<OVALUE>> cache = new HashMap<>();

    /**
     * External value supplier which should be provided if "valuePublisher" with "keyExtractor"
     * are not set
     */

ova2

Flux<String> cars = Flux.just("Fiat", "Audi", "BMW");
Flux<String> years = Flux.just("2009", "2018", "2015");

Flux.zip(cars, years)
    .map(t -> t.getT1() + " " + t.getT2())
    .subscribe(System.out::println);

// Output
Fiat 2009
Audi 2018

ova2

CompletableFuture<Void> cf = CompletableFuture.supplyAsync(() -> 1 + 2)
        .thenApply(i -> "1 + 2 = " + i)
        .thenAccept(s -> System.out.println("Computation result: " + s))
        .thenRun(() -> System.out.println("Finished"));

cf.get();

// Output
Computation result: 1 + 2 = 3
Finished

ova2

@RestController
class TopologyController {
  
    @GetMapping(path = "/uno", produces = {APPLICATION_JSON_VALUE})
    ...
    public Mono<UnoTopologyDto> getUnoTopology(
           @RequestParam Set<String> stationNames,
           @RequestParam(name = "replicationId", required = false) Long replicationId,
           @RequestParam(name = "topoReferenceDate", required = false)
           @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate topoReferenceDate) {

ova2

Do something before task execution ...
21
Do something after task execution ...
37
87
93
23