spring - How to avoid repeating maping operations for similar subscribers in Sping Reactor?

Question

I have one publisher that emits strings and many subscribers that may use the same mapping function for creating models with different filters.

Publisher:

val publisher: Flux<String> = ...

Subscriber#1

val sub1 = publisher.map{veryExpensiveConverter.convert(it)}
                    .filter(it.metric<10)

Subscriber#2

val sub2 = publisher.map{veryExpensiveConverter.convert(it)}
                    .filter(it.metric>5)

Subscriber#3

val sub3 = sub2.map{cheapConverter.convert(it)}
                    .filter(it.metric>8)

Subscriber#4

val sub4 = sub3.map{yetAnotherConverter.convert(it)}
                    .filter(it.metric>80)

In the end I subscribe on all fluxes

Flux.merge(sub1, sub2, sub3, ..., subn)
     .map{//some logic for following data of subscribers}
     .subscribe()

The problem: veryExpensiveConverter is executed several time for the same published record for each subscriber. The execution flow looks

Input1 -> veryExpensiveConverter -> filter1 -> output1
       -> veryExpensiveConverter -> filter2 -> output2
       -> veryExpensiveConverter -> cheapConverter -> filter3 -> output3

I would like too have

Input1 -> veryExpensiveConverter -> filter1 -> output1  
                                 -> filter2 -> output2
                                 -> cheapConverter -> filter3 -> output3

What's a pattern is the most suitable for avoiding execution of the same mapping for each subscriber?

question from:https://stackoverflow.com/questions/65877264/how-to-avoid-repeating-maping-operations-for-similar-subscribers-in-sping-reacto

Answer 1

You can .share() at some level to ensure that every subscription to that shared part only trigger a single subscription above it.

You can also look into .publish().xxx() methods for more advanced auto-triggers (.share() will start its source as soon as the first subscription comes in).

Something like this:

val expensiveDoneOnce = publisher
    .map{veryExpensiveConverter.convert(it)}
    .publish()
    .refCount(2)
val sub1 = expensiveDoneOnce.filter(it.metric < 10)
val sub2 = expensiveDoneOnce.filter(it.metric > 5)