使用项目反应器的 flatMap 和 switchIfEmpty 运算符的问题 [英] Issue with use of project reactor's flatMap and switchIfEmpty operators

问题描述

我有一个依赖于 flatMap() 和 switchIfEmpty() 的反应链的问题.出于某种原因，其中一个 Mono 不发出任何东西......

I have an issue with a reactive chain relying on flatMap() and switchIfEmpty(). For some reason, one of the Mono does not emit anything...

这是调用其他方法的公共处理程序方法:

This is the public handler method calling the others:

//Throws: NoSuchElementException: Source was empty
public Mono<ServerResponse> createUser(ServerRequest serverRequest) {
    Hooks.onOperatorDebug();
    Mono<User> userMono = serverRequest.bodyToMono(User.class);
    return validateUser(userMono)
        .switchIfEmpty(saveUser(userMono))
        .single();
}

这是createUser 调用的第一个方法.请注意，它不是从 switchIfEmpty()(见上文)调用的，它确实发出一个错误(如果有的话).

This is the first method called by createUser. Note that it is not called from a switchIfEmpty() (see above) and it does emit an error if there is any.

private Mono<ServerResponse> validateUser(Mono<User> userMono) {
    return userMono
        .map(this::computeErrors)
        .filter(AbstractBindingResult::hasErrors)
        .flatMap(err ->
            status(BAD_REQUEST)
                .contentType(APPLICATION_JSON)
                .body(BodyInserters.fromObject(err.getAllErrors()))
        );
}

这只是一个辅助方法:

private AbstractBindingResult computeErrors(User user) {
    AbstractBindingResult errors = new BeanPropertyBindingResult(user, User.class.getName());
    userValidator.validate(user, errors);
    return errors;
}

这是 saveUser 方法.它不发出任何结果！！.它是从 switchIfEmpty(见上文)调用的.

This is the saveUser method. It does not emit any result!!. It is called from a switchIfEmpty (see above).

private Mono<ServerResponse> saveUser(Mono<User> userMono) {
    return userMono
        .flatMap(userRepository::save)
        .flatMap(newUser -> status(CREATED)
            .contentType(APPLICATION_JSON)
            .body(BodyInserters.fromObject(newUser))
        );
}

但是，如果我直接调用 saveUser 方法，它会发出一个结果.

However, if I just call the saveUser method directly, it will emit a result.

//Works fine
public Mono<ServerResponse> createUser(ServerRequest serverRequest) {
    Hooks.onOperatorDebug();
    Mono<User> userMono = serverRequest.bodyToMono(User.class);
    return saveUser(userMono) // Compare this to the above version
        .single();
}

任何人都可以帮忙弄清楚为什么 saveUser 方法在从 switchIfEmpty() 调用时不发出任何内容?

Can anyone please help figure out why the saveUser method does not emit anything when called from a switchIfEmpty()?

这是我得到的错误:

java.util.NoSuchElementException: Source was empty
        at reactor.core.publisher.MonoSingle$SingleSubscriber.onComplete(MonoSingle.java:165) ~[reactor-core-3.3.0.RC1.jar:3.3.0.RC1]
        Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Assembly trace from producer [reactor.core.publisher.MonoSingleMono] :
        reactor.core.publisher.Mono.single(Mono.java:3898)
        org.example.contracttestingdemo.handler.UserHandler.createUser(UserHandler.java:55)

推荐答案

这与 flatMap() 或 switchIfEmpty() 直接无关 - 这是因为你尝试使用相同的 Mono 两次:

It's not to do with flatMap() or switchIfEmpty() directly - it's because you're trying to consume the same Mono twice:

Mono<User> userMono = serverRequest.bodyToMono(User.class);
return validateUser(userMono)
    .switchIfEmpty(saveUser(userMono))
    .single();

在上面的例子中，你首先将 userMono 传递给 validateUser()，然后传递给 saveUser()(但通过那个指向 User 已经发出.)

In the above example, you're passing userMono first to validateUser(), and then to saveUser() (but by that point the User has already been emitted.)

如果你想让发布者被多次订阅，并输出相同的结果，你需要通过调用serverRequest.bodyToMono(User.class).cache();来缓存它.

If you want the publisher to be subscribed to multiple times, and output the same result, you'll need to cache it by calling serverRequest.bodyToMono(User.class).cache();.

这篇关于使用项目反应器的 flatMap 和 switchIfEmpty 运算符的问题的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！