Hands-On Reactive Programming in Spring 5

Sharing elements of a stream

With ConnectableFlux, we multicast events for a couple of subscribers. However, we are waiting for subscribers to appear and only then do we start processing. The share operator allows the transformation of a cold publisher into a hot one. The operator behaves in a way that propagates the events that the subscriber has not missed yet for each new subscriber. Let's consider the following use case:

Flux<Integer> source = Flux.range(0, 5)    .delayElements(Duration.ofMillis(100))    .doOnSubscribe(s ->        log.info("new subscription for the cold publisher"));Flux<Integer> cachedSource = source.share();cachedSource.subscribe(e -> log.info("[S 1] onNext: {}", e));Thread.sleep(400);cachedSource.subscribe(e -> log.info("[S 2] ...