How should I use Flux.publish()?

Question

I have a Flux of data that I want to transform both in images and in reports. Because I want to reuse the same source of data, I thought of using thepublish method on Flux and concatenate the results like in the code below:

    @Test
    fun `inside a publish, I can concat multiple fluxes`() {
        data class Data(val d: String)
        data class Image(val i: String)
        data class Report(val r: String)

        val result = Flux.just(Data("some data"))
                .publish { fluxOfData ->
                    val fod = fluxOfData  //.cache()
                    val images = fod.flatMap { Flux.just(Image("my image")) }
                    val reports = fod.flatMap { Flux.just(Report("my report")) }

                    Flux.concat(images, reports)
                }
                .collectList()
                .block()

        Assertions.assertEquals(result, listOf(Image("my image"), Report("my report")))
    }

However, the above code only sends [Image(i=my image)] so it's like the fluxOfData is consumed the first time. However I expected that I would be able to subscribe to fluxOfData multiple times.

When caching the fluxOfData by calling fluxOfData.cache(), the results come out as expected.

So what's the use case of Flux.publish()?
Is there another idiomatic way to transform the data into both images and reports?

user_551406 · Answer

publish() turns the Flux into a hot source (rather than a cold source). This means that multiple subscribers can subscribe at any time and see all elements going forward, but they'll miss all elements emitted beforehand.

So what's the use case of Flux.publish()?

There's plenty of use cases for this - perhaps it's a stream of log messages that are being consumed somehow (and you're only interested in log messages that pop up in real-time, not previous logs.) Perhaps it's a "live update" stream covering a sports match. Perhaps it's a chat room.

When caching the fluxOfData by calling fluxOfData.cache(), the results come out as expected.

Yup, because then you're explicitly caching all the signals emitted, so despite being a hot source, new subscribers can still be replayed all the existing elements in the stream.

Is there another idiomatic way to transform the data into both images and reports?

Consider replay() instead of publish():

Turn this Flux into a hot source and cache last emitted signals for further Subscriber. Will retain an unbounded amount of onNext signals. Completion and Error will also be replayed.

In a more general sense, this would also be good background reading.

How should I use Flux.publish()?

1 Answers