Spring Rector如果不为空，则抛出异常。

Question

与Spring Reactor: How to throw an exception when publisher emit a value?类似，

我的DAO中有一个查找方法java findSomePojo，它返回结果SomePojo。这个查找调用了亚马逊数据库API，而javasoftware.amazon.awssdk.services.dynamodb.model.GetItemResponse具有调用输出。

因此，在我的服务层createSomePojo方法中，我尝试使用hasElement()检查（不确定是否正确使用-我正在尝试和调试）。

基本上： 我想检查是否已经有元素，如果保存是非法的，我就不会调用DAO的保存。所以我需要抛出异常。

假设在DB中已经有SomePojo的记录，我尝试调用服务的create_SomePjo。但是我看到日志中过滤器不起作用，并且在反应堆调用createModel_SomePojo时出现NPE，使我相信即使在检查过滤器之后，它仍然会抛出NPE。

///service SomePjoService it has create_SomePojo, find_SomePojo etc

Mono<Void>  create_SomePojo(reqPojo){

// Before calling DAO 's save I call serivice find (which basically calls DAOs find (Shown befow after this methid)
       Mono<Boolean> monoPresent = find_SomePojo(accountId, contentIdExtn)
                 .filter(i -> i.getId() != null)
                 .hasElement();
       System.out.println("monoPresent="+monoPresent.toString());
       if(monoPresent.toString().equals("MonoHasElement")){
       //*************it comes here i see that***********//
           System.out.println("hrereee monoPresent="+monoPresent);
          // Mono<Error> monoCheck=
                return  monoPresent.handle((next, sink) -> sink.error(new SomeException(ITEM_ALREADY_EXISTS))).then();
       } else {
           return SomePojoRepo.save(reqPojo).then();
       }

}

Mono<SomePojo> find_SomePojo(id){
    return SomePojoRepo.find(id);
}

==============================================================

///DAO : SomePojoRepo.java : it has save,find,delete
Mono<SomePojo> find( String id) {
    Mono<SomePojo> fallback = Mono.empty();
    Mono<GetItemResponse> monoFilteredResponse = monoFuture
        .filter(getItemResponse -> getItemResponse.item().size() > 0&& getItemResponse!=null);
    Mono<SomePojo> result = monoFilteredResponse
        .map(getItemResponse -> createModel_SomePojo(getItemResponse.item()));

    Mono<SomePojo> deferedResult = Mono.defer(() -> result.switchIfEmpty(fallback));
        return deferedResult;
}

我看到Mono上有hasElement()方法，但不确定如何正确使用它。 如果我直接在服务中调用DAO save而不进行所有这些findner检查，可能会出现异常，因为主键约束将处理并抛出异常，然后我可以重新抛出并在服务中捕获，但是如果我想在服务中检查并抛出带有错误代码的异常呢？ 思路是不将响应错误对象传递到DAO层。

Translated text:

我看到Mono类上有hasElement()方法，但是不确定如何正确使用它。如果我在服务中直接调用DAO save方法而没有进行findner检查，可能会出现异常，因为主键约束会处理并抛出异常，然后我可以重新抛出并在服务中捕获，但是如果我想在服务中检查并抛出带有错误代码的异常呢？我的思路是不将响应错误对象传递到DAO层。

- Vivek Misra

---

你确定问题与过滤器 Lambda 表达式中的条件没有关联吗？ 表达式从左到右评估。 我认为你应该首先检查 getItemResponse != null。 - undefined

增加了一些细节 - undefined

将过滤条件更改为： getItemResponse -> getItemResponse != null && getItemResponse.item().size() > 0 - undefined

好的。修复了过滤条件。但是一些hasElement()现在不起作用了。我以为它是工作的，但现在即使对于新请求，它也会进入第一个if条件而不是直接保存。所以现在我总是看到我的自定义异常 :). 猜想我没有正确使用hasElement或者正确处理的方法是什么呢？ - undefined

在这段代码中，create_SomePojo()和createModel_SomePojo()是相同的吗？ - undefined

createModel_SomePojo()只是一个基于Amazon DynamoDB映射的Pojo实用工具。它位于DAO中。它基本上通过将DynamoDB列映射到SomePojo来提取GetitemResponse。为了清晰起见，您可以称其为maptoPojo()。DAO只有save/find/delete方法。 Service具有create_SomePojo和find_SomePojo的签名。 - undefined

Answer 1

尝试使用Hooks.onOperatorDebug()挂钩以获得更好的调试体验。

正确使用hasElement的方法（假设find_SomePojo永不返回null）

Mono<Boolean> monoPresent =  find_SomePojo(accountId, contentIdExtn)
        .filter(i -> i.getId() != null)
        .hasElement();

return monoPresent.flatMap(isPresent -> {
    if(isPresent){
        Mono.error(new SomeException(ITEM_ALREADY_EXISTS)));
    }else{
        SomePojoRepo.save(reqPojo);
    }
}).then();

旁注

对于Mono的定义有一个普遍误解。它不保存任何数据 - 它只是管道的一部分，用于传输信号和通过它流动的数据。因此，代码行System.out.println("monoPresent="+monoPresent.toString());没有意义，因为它只打印了hasElements()修饰符包装的现有管道。该修饰符的内部名称是MonoHasElement，无论其中包含什么（true/false），都将打印出MonoHasElement。

正确的打印信号（以及传输的数据）的方式是：Mono.log()、Mono.doOnEach/next(System.out::println) 或 System.out.println("monoPresent="+monoPresent.block());。请注意第三种方式：它将阻塞整个线程，直到数据被发射，因此只有在知道自己在做什么时才使用它。

带有打印Mono的示例：

   Mono<String> abc = Mono.just("abc").delayElement(Duration.ofSeconds(99999999));

    System.out.println(abc); //this will print MonoDelayElement instantly
    System.out.println(abc.block()); //this will print 'abc', if you are patient enough ;^)
    abc.subscribe(System.out::println); //this will also print 'abc' after 99999999 seconds, but without blocking current thread

- arap

---

是的，我的队友提出了同样的建议。谢谢。 - undefined