I'm stuck with understanding of some aspects of Project Reactor.

I have the following list of components:

1. Validator of input params, returns Mono<Void> or Mono.error()
2. Service saving data to db, returns Mono<Item>
3. Logger for successful actions of an user, returns Mono<Void

A business logic is quite simple: validate params (1), save an item to db (2) and log actions (3). The problem is validator (1) returns Mono.empty() if there are no errors with input data and Mono.error() if input params contain some errors.

I would like to achieve the next things:

1. If validator returns Mono.empty() then continue chain
2. If validator returns Mono.error() then immediately stop processing and throw error which will be handled by exceptionHanlder

I have tried two options:

First with .then(Mono<Item> item) after validation. It allows me to execute saving operation after validation. Given that .then() ignores any errors, I can't rise an exception.

return inputValidator.validateFields(userId, projectId)
            .then(repository.save(item))
            .onErrorMap(RepoException.class, ex -> new UnexpectedError("Failed to save item", ex))
            .subscribeOn(Schedulers.boundedElastic())
            .doOnSuccess(n -> logService.logActivity(new Activity(adminId, n))
                    .subscribe());

Second with .flatMap(Function<Mono<Void>, <Mono<? extends Item> func) after validation. This approach can rise an exception from validator, but I can't execute saving operation because flatMap() doesn't trigger on empty result.

return inputValidator.validateFields(userId, projectId)
            .flatMap(v -> repository.save(item))
            .onErrorMap(RepoException.class, ex -> new UnexpectedError("Failed to save item", ex))
            .subscribeOn(Schedulers.boundedElastic())
            .doOnSuccess(n -> logService.logActivity(new Activity(adminId, n))
                    .subscribe());

It is also important to have access for created object after saving (step 2), because I need to pass it to logger service.

CodePudding user response：

You can't use flatMap because there is no onNext signal - use then instead. Not sure what do you mean by "called" but there is a difference between Assembly and Subscription time in reactive. Publisher you specified in then will not be resolved in case inputValidator.validateFields returns onError signal.

Here is a test for failed validation and as you may see subscription was not triggered

@Test
void saveWasNotCalledIfValidationFailed() {
    var saveMock = PublisherProbe.of(Mono.just("id"));

    var repo = mock(Repository.class);
    when(repo.save())
            .thenReturn(saveMock.mono());

    var res = validateFields()
            .then(repo.save())
            .onErrorMap(IllegalArgumentException.class,
                    ex -> new IllegalStateException("Failed to save item", ex)
            );

    StepVerifier.create(res)
            .expectError(IllegalStateException.class)
            .verify();

    saveMock.assertWasNotSubscribed();
}

private Mono<Void> validateFields() {
    return Mono.error(new IllegalArgumentException("oops"));
}

public static class Repository {
    public Mono<String> save() {
        return Mono.just("id");
    }
}

and here is a test for passed validation

@Test
void saveIsCalledIfValidationPassed() {
    var saveMock = PublisherProbe.of(Mono.just("id"));

    var repo = mock(Repository.class);
    when(repo.save())
            .thenReturn(saveMock.mono());

    var res = validateFields()
            .then(repo.save())
            .onErrorMap(IllegalArgumentException.class,
                    ex -> new IllegalStateException("Failed to save item", ex)
            );

    StepVerifier.create(res)
            .expectNext("id")
            .verifyComplete();

    saveMock.assertWasSubscribed();
}

private Mono<Void> validateFields() {
    return Mono.empty();
}