如何检查 Mono 是不是为空？

Posted 2023-02-16

【中文标题】如何检查 Mono 是不是为空？

【英文标题】：How to check if Mono is empty?如何检查 Mono 是否为空？

【发布时间】：2018-04-24 20:05:16

【问题描述】：

我正在使用 WebFlux 框架使用 Spring Boot 2.0 和 Kotlin 开发应用程序。

我想在保存交易之前检查用户 ID 是否存在。我被困在一个简单的事情上，比如验证 Mono 是否为空。

fun createTransaction(serverRequest: ServerRequest) : Mono<ServerResponse> 
    val transaction = serverRequest.body(BodyExtractors.toMono(Transaction::class.java))

    transaction.flatMap 
        val user = userRepository.findById(it.userId)
        // If it's empty, return badRequest() 
     

    return transaction.flatMap  transactionRepository.save(it).then(created(URI.create("/transaction/" + it.id)).build()) 

有可能做我想做的事吗？

【参考方案1】：

首先让我说我是反应式（java）和这个论坛的新手。 我认为如果单声道为空，您将无法真正检查此代码，因为单声道表示稍后将执行的代码，因此在此代码主体中，您尚不知道它是否为空。这有意义吗？

我刚刚用 Java 写了一些类似的东西，似乎可以工作（但不是 100% 这也是最好的方法）：

    public Mono<ServerResponse> queryStore(ServerRequest request)  

        Optional<String> postalCode = request.queryParam("postalCode");                            

        Mono<ServerResponse> badQuery = ServerResponse.badRequest().build();
        Mono<ServerResponse> notFound = ServerResponse.notFound().build();

        if (!postalCode.isPresent())  return  badQuery; 

        Flux<Store> stores = this.repository
                .getNearByStores(postalCode.get(), 5);

        return ServerResponse.ok().contentType(APPLICATION_JSON)
                .body(stores, Store.class)
                .switchIfEmpty(notFound);

【参考方案2】：

允许检查Flux/Mono是否为空的技术

使用运算符.switchIfEmpty/.defaultIfEmpty/Mono.repeatWhenEmpty

使用提到的运算符，您将能够对 Stream 已完成的情况做出反应，而不会发出任何元素。

首先，请记住，如果没有调用onNext，则根本不会调用.map、.flatMap、.filter 和许多其他运算符。 这意味着在您的情况下，下一个代码

transaction.flatMap 
    val user = userRepository.findById(it.userId)
    // If it's empty, return badRequest() 
 

return transaction.flatMap  transactionRepository.save(it).then(created(URI.create("/transaction/" + it.id)).build()) 

根本不会被调用，如果transaction 为空。

如果在你的流为空的情况下需要处理case，你应该考虑像next这样的操作符，方式如下：

transaction
   .flatMap(it -> 
      val user = userRepository.findById(it.userId)
   )
   .swithIfEmpty(Flux.defer(() -> Flux.just(badRequest())));

实际解决方案

另外，我注意到您从主transaction 创建了两个子流。实际上，下面的代码根本不会被执行：

transaction.flatMap 
    val user = userRepository.findById(it.userId)
    // If it's empty, return badRequest() 
  

并且只会执行最后一个，它是从方法返回的。发生这种情况是因为您没有使用运算符.subscribe(...) 订阅。

第二点，您不能一次订阅相同的请求正文（WebClient 的回复的一种限制）。因此，您需要以下一种方式分享您的请求正文，因此完成的示例将是：

fun createTransaction(serverRequest: ServerRequest): Mono<ServerResponse> 
    val transaction = serverRequest.body(BodyExtractors.toMono(Transaction::class.java)).cache()

    transaction
            .flatMap  userRepository.findById(it.userId) 
            .flatMap  transaction.flatMap  transactionRepository.save(it)  
            .flatMap  ServerResponse.created(URI.create("/transaction/" + it.id)).build() 
            .switchIfEmpty(transaction.flatMap  ServerResponse.badRequest().syncBody("missed User for transaction " + it.id) )

或者更简单的情况，不共享交易流但使用Tuple：

fun createTransaction(serverRequest: ServerRequest): Mono<ServerResponse> 
    val emptyUser = !User()
    val transaction = serverRequest.body<Mono<Transaction>>(BodyExtractors.toMono(Transaction::class.java))

    transaction
            .flatMap  t ->
                userRepository.findById(t.userId)
                        .map  Tuples.of(t, it) 
                        .defaultIfEmpty(Tuples.of(t, emptyUser))
            
            .flatMap 
                if (it.t2 != emptyUser) 
                    transactionRepository.save(it.t1)
                            .flatMap  ServerResponse.created(URI.create("/transaction/" + it.id)).build() 
                 else 
                    ServerResponse.badRequest().syncBody("missed User for transaction " + it.t1.id)
                
            

【讨论】：

第一个解决方案无法编译...你能检查一下吗？此外，没有带有字符串参数的 badRequest 方法。

@voliveira，你能否指出这个例子是用哪种语言编写的，因为我只是试图遵循你的代码约定

@voliveira89 已修复

也不编译！您给出的示例（完全等于我在代码中写的）是有道理的，但是 intellij 在关闭函数的括号中引发了一个错误：“具有块体的函数中需要一个'return'表达式（' ...')"

能否请您指出上面代码 sn-p 中的特定行？

【参考方案3】：

我们可以为此使用 switchIfEmpty 方法

以下示例，我正在检查用户是否存在电子邮件，如果不存在则添加它

userRepository.findByEmail(user.getEmail())
                .switchIfEmpty(s -> 
                    user.setStatus("InActive");
                    String encodedPassword = DigestUtils.sha256Hex(user.getPassword());
                    user.setPassword(encodedPassword);
                    userRepository.save(user).subscribe();
                    s.onComplete();
                ).then(Mono.just(user));

【参考方案4】：

您可以使用 Mono 提供的方法 hasElement() 进行检查，该方法类似于 Optional 的 isPresent()。方法定义为：

Mono<Boolean> hasElement()

更多详情请查看：project reactor documentation

如果您必须基于此值执行某些操作，您可以进一步使用 switchIfEmpty() 来提供备用发布者。