Hands-On Reactive Programming in Spring 5

Caching elements of a stream

With ConnectableFlux, it is easy to implement different data caching strategies. However, Reactor already has the API for event caching in the form of the cache operator. Internally, the cache operator uses ConnectableFlux, so the primary added value of it is a fluent and straightforward API. We may tune the amount of data our cache can hold and the expiration time for each cached item. Let's demonstrate how it works with the following example:

Flux<Integer> source = Flux.range(0, 2)                              // (1)    .doOnSubscribe(s ->        log.info("new subscription for the cold publisher"));Flux<Integer> cachedSource = source.cache(Duration.ofSeconds(1));    // (2)cachedSource.subscribe(e -> log.info("[S 1] onNext: {}", e));        // (3)cachedSource.subscribe(e ...