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

ova2

    public <T> Mono<T> executeMono(Supplier<T> task) {
        CompletableFuture<T> future = CompletableFuture.supplyAsync(task);
        return Mono.fromFuture(future);
    }

    public <T> Flux<T> executeFlux(Supplier<List<T>> task) {
        return Flux.create((sink) -> {
            CompletableFuture<List<T>> future = CompletableFuture.supplyAsync(task);
            future.whenComplete((list, exception) -> {

ova2

String result = CompletableFuture.supplyAsync(() -> "1 + 2 = " + (1 + 2)).get();

System.out.printl(result);

// Output
1 + 2 = 3

ova2

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;

class Test {