how to handle multiple subscribe to a Uni

[liuzqt]

This is not a issue/bug, it's a usage question but I really can not figure it out after reading through the whole guide....

let's say we create several Unit from a single upstream Uni, and all of them subscribe, which will trigger the upstream unit multiple times, like the exmaple below

        AtomicInteger counter = new AtomicInteger();
        Uni<String> upstream = Uni.createFrom().item(() -> String.valueOf(counter.getAndIncrement()));
        Uni<String> sub1 = upstream.onItem().transform(i -> "got item" + i + " and perform transform 1");
        Uni<String> sub2 = upstream.onItem().transform(i -> "got item" + i + " and perform transform 2");
        sub1.subscribe().with(System.out::println);
        sub2.subscribe().with(System.out::println);

What I want is to trigger upstream unit only one, when materializing the path, when you find the Uni node already 'trigger', just fetch that value. It's common that you want to subscribe from an asynchronous upstream and then perform different downstream task in totally different branches, just like a split DAG

[cescoffier]

You are looking for memoize:

Uni<String> upstream = Uni.createFrom().item(() -> String.valueOf(counter.getAndIncrement()))
        .memoize().indefinitely();

[liuzqt]

@cescoffier thanks that's exactly what I'm looking for. But BTW why we're calling indefinitely() here, or in another word, why memorize do not produce another Uni? I feel like it has no different from transform or transformToUni, just 'mark' a Uni as 'do not trigger again'.

I've come up with another walkaround previously, which is to await blocking until we got the result from that Uni and then perform different tasks on that intermidiate result, it seems that this approach is equivalent to memorize?

[cescoffier]

memoize does not block and orchestrates the subscription request to have only one. So it's much better than the blocking approach.

memoize().indefinitely() returns a Uni, and the emitted value is cached indefinitely (so it's not the same thing as "await().indefinitely(). You can also decide to cache the value for a period of time (memoize(). until(...)`)

[liuzqt]

@cescoffier got it. It's much more clear to me now. Thanks for the explanation!