Reactor-core: add .delayCompletionUntil / or .concatWithEmpty operator for Flux/Mono

[62mkv]

Motivation

There're use cases when we have a Publisher, and need to produce another Publisher that will emit exactly the same items, as the source, but will also trigger completion of the other Publisher.

Desired solution

Flux<Item> source;

source
   .flatMap(this::process)
   .delayCompletionUntil(commitTransaction());

Considered alternatives

Currently (with assistance from @bsideup) I ended up with:

Flux<Item> source;

source
   .flatMap(this::process)
   .concatWith(commitTransaction().ignoreElement().cast(Item.class));

those .ignoreElement() / .cast() feel hacky here. We have a clear intention - emit exactly same items, as a source, but before completion, perform some other reactive action (which is impossible to do in doOnComplete, because it's not being subscribed to)

[bsideup]

@62mkv I just updated the code samples to make them more focused on the issue, I hope you don't mind.

[simonbasle]

There's another alternative by putting then on the commitTransaction() flux:

Flux<Item> source;

source
   .flatMap(this::process)
   .concatWith(commitTransaction().then(Mono.empty());

wdyt?
I'm not terribly in favor of adding an operator for that use case, which seems a bit niche... And there's not that much boilerplate to write, even though finding _what_ to write is the challenge here 😄 Maybe worth an entry in the FAQ ?

[62mkv]

@simonbasle from what I've learned thus far, .concatWith will not accept publisher without casting.. which will create another problem for any noob (like myself)

[simonbasle]

the alternative solution I provide removes the need for casting, as the compiler understands that Mono.empty()'s generic type must be and can be the same as that of the source Flux.