How to throw an exception properly when do Flux processing?

Question

Existing code that I have:

private Flux<Integer> testGetFluxTestData() {
    return Flux.just(new TestData(1), new TestData(2))
            .collectList()
            .map(list -> list.stream()
                    .map(TestData::getId)
                    .collect(Collectors.toList()))
            .flatMapMany(Flux::fromIterable);
}

I want to enrich existing code and throw an exception when some not allowed data received, I made the following changes:

    private Flux<Integer> testGetFluxTestData2() {
        return Flux.just(new TestData(1), new TestData(2))
                .collectList()
                .map(list -> {
                    return !list.contains(new TestData(1)) ?
                            list.stream()
                                    .map(TestData::getId)
                                    .collect(Collectors.toList()) :
                            Flux.error(new IllegalTestDataException("illegal test data 1"));
                })
                .flatMapMany(Flux::fromIterable);
    }

but my implementation even noncompilable due to the following line:

Flux.error(new IllegalTestDataException("illegal test data 1"));

Could you please suggest, how to handle exception throwing for my particular scenario?

Does this answer your question? [Correct way of throwing exceptions with Reactor](https://stackoverflow.com/questions/53595420/correct-way-of-throwing-exceptions-with-reactor) — Oleh Dokuka, Oct 05 '20 at 13:20

score 1 · Accepted Answer · answered Oct 06 '20 at 07:33

You are attempting to map from a List<TestData> to either a List<Integer> or a Flux<?> (error), which makes the desired result type ambiguous. Returning a reactive type in a mapping function is generally not desired (you'd want to do that in a flatmapping function).

(side note: even if you were in a flatMap, it wouldn't work either because at that point you're in Mono API due to collectList, so Mono.flatMap expects a Mono result to the Function).

Note that the map operator catches exceptions from the lambda and turn them into an onError signal, so technically you could replace the Flux.error with a throw.

Otherwise, you'd need to turn the map into a flatMap and the Flux.error into a Mono.error, for the reasons stated above.

How to throw an exception properly when do Flux processing?

1 Answers1