How to repeat Mono while not empty

Question

I have a method which returns like this!

Mono<Integer> getNumberFromSomewhere();

I need to keep calling this until it has no more items to emit. That is I need to make this as Flux<Integer>.

One option is to add repeat. the point is - I want to stop when the above method emits the first empty signal.

Is there any way to do this? I am looking for a clean way.

Answer 1

A built-in operator that does that (although it is intended for "deeper" nesting) is expand.

expand naturally stops expansion when the returned Publisher completes empty.

You could apply it to your use-case like this:

//this changes each time one subscribes to it
Mono<Integer> monoWithUnderlyingState;

Flux<Integer> repeated = monoWithUnderlyingState
    .expand(i -> monoWithUnderlyingState);

Answer 2

I'm not aware of a built-in operator which would do the job straightaway. However, it can be done using a wrapper class and a mix of operators:

Flux<Integer> repeatUntilEmpty() {
    return getNumberFromSomewhere()
        .map(ResultWrapper::new)
        .defaultIfEmpty(ResultWrapper.EMPTY)
        .repeat()
        .takeWhile(ResultWrapper::isNotEmpty)
}

// helper class, not necessarily needs to be Java record
record ResultWrapper(Integer value) {
    public static final ResultWrapper EMPTY = new ResultWrapper(null);

    public boolean isNotEmpty() {
        return value != null;
    }
}