Reactor-core: `firstValue()` variant for `Mono` and `Flux`

[berry120]

Mono and Flux provide a first() method used for forming a publisher that replays all the signals of whichever publisher responds with any signal the fastest (as the docs say, effectively behaving like the fastest of the competing sources.) There are also or() methods which have the same behaviour, but are instance rather than static methods.

However, no such method exists for replaying whichever publisher that emits the first value, rather than just the first signal.

Motivation

This is useful when wanting to use the fastest of several sources, but ignoring any sources that are empty or return an error.

For instance, one may wish to use a value / stream of values returned by the fastest of a number of distributed sources / services; but would not want to consider any empty or erroneous sources.

Desired solution

Mono.firstValue() and Flux.firstValue() methods that use the publisher which returns the first value, not the first of any signal.

Considered alternatives

Mono.firstValue() could essentially be replaced by using Flux.merge(mono0, mono1).next();. Flux.firstValue(). Not immediately clear if there's an obvious way of emulating the same behaviour with a Flux.

[simonbasle]

I couldn't find an existing duplicate issue, but I'm pretty sure this has been brought up in the past 🤔
I agree that it could be useful, especially as people tend to expect the empty case to be ignored by or.

I'm just sad that first and or are so perfect in terms of names for discoverability of whichever use case we want to promote the most: I don't want to simply deprecate these methods nor ultimately get rid of them 😢 So we need to be careful about a migration path, or how we communicate that change to users.

That said, we _could_ take the opportunity to change the current behavior in 3.4, so that:

• first and or would now consider only the (first) onNext signal for choosing the winning source
• losing source is still cancelled
• if no source emits an onNext we have a few choices: see below
• we'd provide the exact same old behavior under new operators like firstSignal and orSignal

wdyt? cc @reactor/discuss, #rstoyanchev

[simonbasle]

Regarding what to do when no source emits onNext: currently, since this case isn't "special" it boils down to propagating whichever terminal signal comes first in the timeline. But with a more data-driven first, maybe we want to do things a little differently?

1. If both sources complete empty, should that be an error (like NoSuchElementException for instance)?

2. Now say one source completes and the other errors. Should the error be taken into account? Or should we consider that, since one of the sources terminated "successfully", the whole sequence is successful (albeit empty)? (if we answered yes to point 1) then probably not, the error would "win" over an empty source)

3. If both sources error, I think we can safely say that the whole sequence errors. Again there is a choice though: should both error be propagated as a composite, or should one of the errors be dropped in favor of the other?

[bsideup]

Re naming - what about Mono.race?

[berry120]

@simonbasle I believe this came up on a SO question a while back - that may be what you're thinking of? Though I can't find that now either...

we'd provide the exact same old behavior under new operators like firstSignal and orSignal

Totally agree that first() and or() are great names - but as you say, in my experience the generally expected behaviour I see others assume is that they're based on values, not signals, which leads to some confusion. (If one wanted them to be based on signals, one could always materialize() the stream first reasonably trivially.) So if this was a new API, then I'd say the behaviour should definitely be value based and the existing variants should be dropped or renamed as above - but I'm always very hesitant to recommend a breaking change in between versions that won't be immediately obvious (ie. won't cause a compilation failure.)

Perhaps the methods could be removed entirely (or renamed) for a major release, before re-introducing them with the new behaviour in a next major release? Seems a bit of a "hack", but the only way I can think of to keep the name without potentially introducing undetected breaking changes.

As to the specific points you raise - I'd say:

1. Makes no sense to use a value-based method to never retrieve a value, so I'd say that should indeed error with a NoSuchElementException.

2. I would actually say we should error with NoSuchElementException any time we don't get a result from any inner publisher, whatever reason that's for (error or empty completion.) This keeps things consistent and means we don't have to introduce any special rules around error handling in separate cases.

3. (Same as point 2 above - even if everything errors, I'd still vote for a NoSuchElementException.)

We could of course populate more specific information on what errored in a cause to the NoSuchElementException, or indeed just the exception message itself.

@bsideup I quite like the idea of Mono.race as a potential alternative name.

[simonbasle]

@berry120 good idea on consistently propagating NoSuchElementException ! The refined cause could easily be detectable in the getCause():

• null cause means ALL sources completed empty
• A non-composite cause means ONE source errored while the rest didn't produce any value
• A composite cause means AT LEAST two sources errored, while the rest didn't produce any value

On the other hand, if ANY source produces a value then this value would "win" over an error (so error(s) in remaining sources are ignored). I feel this would be satisfactory, as there's always the possibility of doOnError if one wants to log the "losing" error(s). wdyt ?

[simonbasle]

@bsideup @berry120 regarding naming, race sounds okay-ish (it tends to make me think of race conditions). I was also thinking about compete. I still prefer first, and there's the question of the instance method (where, again or seems perfect).

[berry120]

@simonbasle I think ignoring errors (in the case where one publisher completes) is fine, since the point of the method is to explicitly ignore any sources that don't publish a value (whether that's because they error, or complete without publishing anything.)

Another advantage for this behaviour is that not ignoring errors makes the result potentially time-dependent. If we did somehow abort or change the result somehow on an error, then given two publishers, one of which errors and one that publishes a value:

• If the successful one returns first (because that "finishes" the race first so to speak) the other is ignored, and so the error would never be propagated.
• If the "errored" publisher returns first then the successful one would be ignored and an error propagated.

Given that we always want the value, that's a bit of an undesirable situation IMHO. I would therefore say erroneous publishers should always be ignored (assuming another publisher that does return a value successfully of course.) As you say, doOnError() can still be used for logging if necessary.

Absolutely agree with the composite exceptions as a cause to NoSuchElementException too.

[berry120]

@bsideup @simonbasle Another contender on the naming front - what about Mono.fastest() or Mono.fastestValue()?

Boring as it is, I think firstValue() is actually my favourite thus far though. The original "signal" behaviour could then be kept for first() as-is.

[Tandolf]

had to chip in, my vote goes to Mono.winner() and Mono.winningValue() everything in life is a competition :)

[bsideup]

FTR here is a workaround for the current API:

List<Flux<Integer>> sources = Arrays.asList(
        Flux.empty(),
        Flux.just(1, 2, 3)
);

StepVerifier.create(Flux.first(sources))
            .verifyComplete();

Flux<Integer> result = Flux.defer(() -> {
    AtomicInteger counter = new AtomicInteger(sources.size());
    DirectProcessor<Integer> processor = DirectProcessor.create();

    return Flux.first(
            sources.stream()
                   .map(it -> {
                       return it
                               .doFinally(s -> {
                                   if (counter.decrementAndGet() == 0) {
                                       processor.onComplete();
                                   }
                               })
                               .mergeWith(processor);
                   })
                   .collect(Collectors.toList())
    );
});

StepVerifier.create(result)
    .expectNext(1, 2, 3)
    .verifyComplete();

[simonbasle]

reviewing this, I find firstValue a tad too confusing. @aneveu used firstValues for the Flux variant so as not to confuse users into thinking this takes the first element and ignores the subsequent elements, but I think firstValued for both Flux and Mono variant would work even better.

I also wonder if we should alias first as firstSignalling or something like that, and deprecate the first altogether... But that's for a separate issue.

Another follow up issue is what to do about or. In the current PR, having an instance method introduces more confusion than what it solves, I think, so we'll delay that and just add the static operator for now.

[simonbasle]

sounds like valued doesn't really convey the meaning I intended of "which has a defined value", so now I'm thinking about firstWithValue. What do you think? It is longer but also way less ambiguous/confusing.

[ericbottard]

How was valued (mis)interpreted?

In an ideal world, the word "next" would appear (because that's the "name" of the signal, but seems on its own it rarely appears. Only rings a bell in the construct "onNext")

firstWithValue is a good compromise though.

[simonbasle]

as I understand for a native speaker, valued means appreciated rather than defined/non-empty so it can lead to confusion.

he other difficulty in naming this operator is to avoid conveying the impression that only the first onNext of the the "winning" source is propagated (which is kind of the case with firstValue, as well as variations like firstOnNext).

I liked first, but maybe as @bsideup suggested a radically different word would be good, like compete. In that case though, I think the old first and or methods should be deprecated and replaced with a compete-based alias. I'm thinking competeOnSignals, competeOnSignalsWith, competeOnValue[s], eventually competeOnValue[s]With

_oops we can't do a vote with reaction emojis since github is limiting the emojis_

[simonbasle]

any feedback on firstWithValue vs competeOnXxx (which kinda allows the instance methods by appending the With suffix) ?

cc @rstoyanchev @bsideup @berry120 @Tandolf @reactor/core-team

[bsideup]

maybe as @bsideup suggested a radically different word would be good, like compete

Or... Mono.race, as @bsideup suggested 😅

[berry120]

My vote goes to firstWithValue.

I'm not keen on compete - it isn't clear that the publishers competing to be the fastest - they could instead be competing to have the best result. My initial reaction when hearing competeOnValue was that (to use a trivial example), given two Publisher<Integer>, the one returning the greater integer might "win", rather than the one returning an integer first.

race is clearer in that regard, but it's not my favourite choice as it smells too much of concurrency / race hazards.

[simonbasle]

race is clearer in that regard, but it's not my favourite choice as it smells too much of concurrency / race hazards

I agree, so let's forget about that one.

[camsteffen]

How about selectFirst or selectFirstWithValue? Otherwise I like firstWithValue.

[simonbasle]

ok so there seem to be a general consensus with firstWithValue. no need to make it plural in Flux I think: "first of the given flux with a value", the with lifts the ambiguity that we were encountering previously.

So let's add static Flux.firstWithValue and Mono.firstWithValue 👍

Three pending questions:

• should we aim at adding an equivalent to or? (can be the focus of a follow-up issue, it's bound to lead to protracted discussions around the naming again)
• should we alias first (eg. as firstWithSignal)?
• should we deprecate first (and or?), and if so with a goal of:
  
  
  
  • swapping their behavior in 3.5 to the firstWithValue one?
  
  
  • simply deleting them in 3.5?
  
  

[ericbottard]

_Reposting as 4 separate comments for gathering votes._

should we aim at adding an equivalent to or? (can be the focus of a follow-up issue, it's bound to lead to protracted discussions around the naming again)

[ericbottard]

should we alias first (eg. as firstWithSignal)?

[ericbottard]

should we deprecate first (and or?), and if so with a goal of swapping their behavior in 3.5 to the firstWithValue one?

[ericbottard]

should we deprecate first (and or?), and if so with a goal of simply deleting them in 3.5?

[berry120]

Few thoughts around the 3 issues mentioned:

should we deprecate first (and or?), and if so with a goal of:

• swapping their behavior in 3.5 to the firstWithValue one?
• simply deleting them in 3.5?

Swapping behaviour is a bad idea IMHO. I'm all for compile-time breaking changes between releases, but this would be invisible at compile-time, and thus could cause some rather unsubtle and hard-to-detect bugs for those relying on the current behaviour. (In an ideal world people would move away from the deprecated method and then could re-introduce it with the new behaviour if desired, but we all know how that works in reality...!)

As for deprecating / removing it - I'm undecided here. first is a nice concise name, and in the case where the first signal really is desired then it reads very nicely. But it is ambiguous - and if we're aliasing the current first behaviour anyway, it's also surplus to requirements. I think I'd lean towards deprecation and removal based on those points, but I don't have a particularly hard opinion on that either way.