spring - Spring Reactive：服务方法调用无法返回响应

Question

试图通过在两者之间包含一个小的验证来重写一个有效的反应函数。不幸的是，即使在后端完成了计算，程序也无法完成请求。此外，如果有人能告诉我是否有任何与响应式相关的不必要的编码实践，那就太好了。

1. 处理函数，我们从这里调用响应式服务

       public Mono<ServerResponse> performValidation(ServerRequest serverRequest) {
       Mono<Request> request = serverRequest.bodyToMono(Request.class);
       return ServerResponse.ok()
               .body(service1.performValidation(boundaryCheckRequest, ""), ServiceResponse.class)
               //.body(service1.performValidation_Working(boundaryCheckRequest, ""), ServiceResponse.class)
               .onErrorResume(Exception.class,
                       e -> ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR)
                               .body(Mono.just(e.getMessage()), String.class));}
   

2. 处理程序调用下面的服务，下面的代码片段有问题

   public Mono<ServiceResponse> performValidation(Mono<ServiceRequest> requestMono,String token) {
       return Mono.create(monoSink -> {
           requestMono.map(triggerRequest -> {
               return checkValidOrders(triggerRequest)?doValidations(requestMono,token).subscribe():Mono.just(new ServiceResponse()).subscribe();
           }).subscribe();
       });
   }
   
   private Mono<ServiceResponse> doValidations(ServiceRequest request,String token){
       RequestHolder requestHolder= new RequestHolder(request,token);
       ServiceResponseObject serviceBus=new ServiceResponseObject();
       return Mono.create(monoSink -> {
           Flux.merge(
                   getServiceOne(requestHolder, serviceBus),
                   getServiceTwo(requestHolder, serviceBus)
                 )
                   .doOnNext(completedFlag -> {
                   })
                   .doOnComplete(() -> {
                       ServiceResponse ServiceResponse = validationService.validate(serviceBus, requestHolder.getRequest());
                       monoSink.success(ServiceResponse);
                   })
                   .doOnError(throwable -> {
                       monoSink.success(handleError(throwable, requestHolder));
                   })
                   .subscribe();
       });
   }
     private Boolean checkValidOrders(ServiceRequest tr){
       return tr.getLines().stream().anyMatch(line-> !line.getStatus().equals("ACIVE"));
   }   
   

3. 服务代码的工作版本。

    public Mono<ServiceResponse> performValidation_Working(Mono<ServiceRequest> requestMono,String token) {
        return Mono.create(monoSink -> {
            requestMono.map((triggerReq)->
            {
                return new RequestHolder(request,token);
            })
            .subscribe(servicesRequestHolder -> {
                ServiceResponseObject serviceBus=new ServiceResponseObject();
                Flux.merge(
                        getServiceOne(requestHolder, serviceBus),
                        getServiceTwo(requestHolder, serviceBus)
                )
                    .doOnNext(completedFlag -> { })
                    .doOnComplete(()->{
                        ServiceResponse ServiceResponse = validationService.validate(serviceBus, requestHolder.getRequest());
                        monoSink.success(ServiceResponse);
                    })
                    .doOnError(throwable -> {
                        monoSink.success(handleError(throwable,servicesRequestHolder));
                    })
                    .subscribe();
            });
        });
    } 

无法弄清楚为什么请求处理没有结束，请告知。

标签： springjava-8reactive-programmingspring-webfluxspring-reactive