spring数据couchbase，无法使用反应编程按id删除文档

Question

最近我们决定在我们的项目中使用spring-WebFlux和couchbase，并且我们需要帮助如何在反应编程中解决下面的用例

>

验证请求并将其保存在Bucket1 couchbase中（我们使用了javax.validation和spring的ReactiveCouchBaseRepository。

调用外部服务（我们使用webclient调用API。

>

成功后,

• 将审核文档写入bucket2。
• 获取插入到Bucket1中的文档并发送该文档作为响应。
• 将审核文档写入bucket2

一旦失败,

• 将审核文档写入bucket2。
• 删除插入到BUCKET1中的文档并引发异常。
• 将审核文档写入bucket2

我们编写了一个服务类，并使用两个存储库类将文档保存到couchbase，以及一个webclient调用外部服务。

我们的服务类方法业务逻辑如下所示。

    {

    //1. Validate the request and throw the error
    List<String> validationMessages = handler.validate(customerRequest);
    if (validationMessages != null && !validationMessages.isEmpty()) {
        return Mono.error(new InvalidRequestException("Invalid Request", validationMessages, null));
    }

    //generate the id, set it to the request and save it to BUCKET1
    String customerRequestId = sequenceGenerator.nextId(Sequence.CUSTOMER_ACCOUNT_ID);
    customerRequest.setcustomerRequestId(customerRequestId);
    customerRequestMono = bucket1Repository.save(customerRequest);


    //2. Call the external service using webclient
    externalServiceResponse = customerRequestWebClient.createCFEEnrollment(customerRequest);

    //2. Subscribe to the response and and on Success write audit to BUCKET2 , and onerror write audit to BUCKET2 , and delete the inserted documet from BUCKET1
    externalServiceResponse.subscribe(response -> {
        //Initialise the success audit bean and save
        //2.1 a) Write Audt to BUCKET2
        Mono<CustomerAuditBean> auditResponse = bucket2Repository.save(cfeAudit);
         }, errorResp -> {
        //2.2 a) Write Audt to BUCKET2
        //Initialise the error audit bean and save
        Mono<CustomerAuditBean> auditResponse = bucket2Repository.save(cfeAudit);

        //2.2 b)Delete the inserted
        Mono<CustomerRequest> delCustomer = bucket1Repository.deleteByLoanAccountId(loanAccountId);
    });

    //Get the loan account id and return the same
    finalResponse = bucket1Repository.findByCustomerId(customerId);
    return Mono.when(externalServiceResponse,customerRequestMono,finalResponse).then(finalResponse)
            .doOnSuccess(resp -> {
                try {
                    finalMasterAudit.setServiceResponse(new ObjectMapper().writeValueAsString(resp));
                    Mono<CustomerAuditBean> auditResponse = bucket2Repository.save(finalMasterAudit);
                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }

            })
            .doOnError(error -> {
                try {
                    finalMasterAudit.setServiceResponse(new ObjectMapper().writeValueAsString(error.getMessage()));
                    Mono<CustomerAuditBean> auditResponse = bucket2Repository.save(finalMasterAudit);
                } catch (JsonProcessingException e) {
                    e.printStackTrace();
                }

            });
}

我们观察者的两个问题

1. 文档在某些情况下是不会持久化的，直到我们订阅它。这是预期的行为吗？是否需要订阅要保存的文档？
2. 出错时无法删除文档。
3. 我也知道我没有遵循上面的纯反应编程。帮助我使用任何指针来有效地以反应方式编写代码。

请帮我们指点一下

Answer 1

以下是上面的一段代码：

externalServiceResponse.subscribe(response -> {
    Mono<CustomerAuditBean> auditResponse = bucket2Repository.save(cfeAudit);
     }, errorResp -> {
    Mono<CustomerAuditBean> auditResponse = bucket2Repository.save(cfeAudit);
    Mono<CustomerRequest> delCustomer = bucket1Repository.deleteByLoanAccountId(loanAccountId);
});

它有两个反应性编程问题：

1. 您创建的是不订阅的单声道，因此它们永远不会执行。
2. 无论如何都不应该在subscribe中创建它们，而是使用flatMap或onErrorResume将它们链接起来，进行预订阅。

类似这样的东西应该能起到作用（请原谅，我还没有测试过它，所以您可能需要做一些调整）：

externalServiceResponse
   // If something goes wrong then delete the inserted doc
   .onErrorResume(err -> bucket1Repository.deleteByLoanAccountId(loanAccountId))

   // Always want to save the audit regardless
   .then(bucket2Repository.save(cfeAudit))

   .subscribe();

代码中还有其他问题需要解决，例如，在订阅最终的单声道之前，您似乎想要将多个单声道平面映射在一起，但希望这能让您开始。