反应式编程-Webflux Webfilter运行不正常

Question

我对反应式编程有点陌生，我正在尝试组装以下组件：使用Java、Springboot 2、Webflux和reactor core，我想处理需要额外身份验证的非常特定的请求。因此，我通过一系列步骤实现了一个Web过滤器：

• 捕获请求的路径和方法。检查该组合是否存在，是否需要使用accessPointService进行特定身份验证。getAccessPointAuthorizationRequirement方法（返回带布尔值的Mono）
• 因为我配置了CSRF和Spring Security性，所以我需要CSRF令牌和springsession凭据。我对凭据发出GET和POST请求
• 然后使用凭据，我只需向一个服务（authcheck）发出POST请求，该服务可以执行一系列安全检查（该服务正常，从Postman和Angular可以正常工作）
• 之后，我需要检索主体，将其转换为字符串，并对其进行检查。现在这种情况没有发生

过滤器

@Override
    public Mono<Void> filter(final ServerWebExchange serverWebExchange, final WebFilterChain webFilterChain) {

        //client for specific requests.
        WebClient webClient = WebClient.builder()
                .baseUrl("http://localhost:8080")
                .build();
        //get request for the CSRF cookie.
        WebClient.RequestHeadersSpec<?> getRequest = webClient.get()
                .uri("/login");
        //post request for the spring security session cookie.
        WebClient.RequestHeadersSpec<?> postRequest = webClient.post()
                .uri("/login")
                .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_FORM_URLENCODED_VALUE)
                .body(BodyInserters.fromFormData("username", "username")
                        .with("password", "password"));
                        //services that checks if the given request needs extra authentication
        return accessPointService.getAccessPointAuthorizationRequirement(serverWebExchange.getRequest().getMethod().toString().toUpperCase(), serverWebExchange.getRequest().getPath().toString())
                .log()
                //gets the csrf token from the GET request
                .flatMap(isRequired -> getRequest.exchangeToMono(response -> Mono.just(response.cookies().getFirst("XSRF-TOKEN").getValue())))
                //combines the previous token with the POST request SESSION cookie,
                //THEN secures the last request with both credentials
                .zipWith(postRequest.exchangeToMono(resp -> Mono.just(resp.cookies().getFirst("SESSION").getValue())),
                        AuthenticationFilter::secureAuthRequest)
                //gets the exchange from the request and converts the body into a String
                .flatMap(AuthenticationFilter::getRequestExchange)
                //code to validate if it's doing something. Not implemented yet because it never executes.
                .flatMap(s -> Mono.just(s.equals("")))
                .onErrorResume(e -> {
                    throw (CustomException) e;//breaks the execution
                })
                .then(webFilterChain.filter(serverWebExchange));//continues the execution
    }

调用了secureAuthRequest和getRequestExchange方法

//adds the springsession cookie and csrf cookie to the request
private static WebClient.RequestHeadersSpec<?> secureAuthRequest(String csrf, String spring) {

        WebClient webClient = WebClient.builder()
                .baseUrl("http://localhost:8080")
                .build();
        WebClient.RequestHeadersSpec<?> request = webClient.post()
                .uri("/authcheck")
                .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
        request.header("X-XSRF-TOKEN", csrf);
        request.cookies( cookies -> cookies.add(  "XSRF-TOKEN", csrf) );
        request.header("Authorization", spring);
        return request;
    }

//gets the body as string.
private static Mono<String> getRequestExchange(WebClient.RequestHeadersSpec<?> securedReq) {

        return securedReq.exchangeToMono(clientResponse -> clientResponse.bodyToMono(String.class));
    }

但是，当请求被绑定为要进行身份验证时，日志如下所示：

2021-10-26 23:57:18.760  INFO 6860 --- [ctor-http-nio-4] reactor.Mono.Just.4                      : | onSubscribe([Synchronous Fuseable] Operators.ScalarSubscription)
2021-10-26 23:57:18.761  INFO 6860 --- [ctor-http-nio-4] reactor.Mono.Just.4                      : | request(unbounded)
2021-10-26 23:57:18.761  INFO 6860 --- [ctor-http-nio-4] reactor.Mono.Just.4                      : | onNext(true)
2021-10-26 23:57:18.762  INFO 6860 --- [ctor-http-nio-4] reactor.Mono.Just.4                      : | onComplete()

据我所知，数据流从一个订阅和一个后请求开始（我认为从accessPointService.getAccessPointAuthorizationRequirement方法Mono值返回TRUE，如果我错了，请更正），但随后会显示“onComplete（）”日志。我不知道onComplete（）日志的确切含义，因为它是在执行getRequestExchange方法（调用该方法）之前显示的。单声道。只有（s.equals（“”）一段代码永远不会执行。

我读过很多关于“订阅之前什么都不会发生”的书，但我仍然不知道如果我从未明确订阅流，为什么会调用反应流，也不知道如何实现它，因为它只返回一次性的（我想我可以从内部抛出异常？）。此外，我听说在调用多个订阅者时要进行解耦，所以我尽量避免使用它们。

任何关于反应性编程、Reactor堆芯或特定流程以及如何改进的帮助，我们都将不胜感激。

干杯

Answer 1

因此，在进行了一些研究之后，感谢@Toerktumlare的评论，并弄清楚了发生了什么，以及我对此做了哪些更改/应用。

因此，对于“onComplete（）”日志，它标志着数据生产者的结束。因此，为了查看操作的完整堆栈，我需要将每个生产者与自己的日志链接起来。例如：

Mono.just(Boolean.FALSE)
    .log()
    .flatMap(booleanVal -> Mono.just(booleanVal.toString()))
    .log()
    .subscribe(stringVal -> System.out.println("This is the boolean value " + stringVal));

这将为初始生产者和平面图操作生成跟踪。

现在，关于主要问题，问题出现在getRequestExchange方法中：

//gets the body as string.
private static Mono<String> getRequestExchange(WebClient.RequestHeadersSpec<?> securedReq) {

        return securedReq.exchangeToMono(clientResponse -> clientResponse.bodyToMono(String.class));
    }

问题隐藏在BodyTomino方法中。根据本网站https://medium.com/@Jeevjyotsinghchabda/dont-let-webclient-s-bodytomono-trick-you-645123b3e0a9，如果此请求的响应由于任何原因没有正文，则不会抛出任何错误，只需返回一个Mono。空（）。因为流程没有为这样的生产商做好准备，所以就在这里结束了。

在我的例子中，问题是spring云安全。我在请求中提供了授权凭据，但没有提供关联的会话cookie。因此，请求返回了一个302（找到），没有正文。这就是问题所在（不是反应流本身）。

所以，在那之后，我修改了请求，@Toerkumlare的评论帮助我开发了一个工作解决方案：

//service that returns if certain resource needs authentication or not, or if it's not even configured
return accessPointService.getAccessPointAuthorizationRequirement(serverWebExchange.getRequest().getMethod().toString().toUpperCase(), FWKUtils.translateAccessPointPath(serverWebExchange.getRequest().getPath().pathWithinApplication().elements()))
                //if the response is a Mono Empty, then returns a not acceptable exception
                .switchIfEmpty(Mono.defer(() -> throwNotAcceptable(serverWebExchange)))
                //takes the boolean value to check if extra auth is needed.
                .flatMap(isRequired -> validateAuthenticationRequirement(isRequired))
                //gets the access token - the extra auth credential
                .flatMap(isRequired -> getHeaderToken(serverWebExchange))
                //from this access generates a WebClient to the specific authentication service - from a webClientProvider to not create too many WebClients.
                .flatMap(accessToken -> generateAuthenticationRequest(webClientProvider.getInstance(), accessToken))
                //gets the CRSF token credential and secures the request (adds it to the header and the cookies)
                .zipWith(getCredential(webClientProvider.getInstance(), "csrf"), (securedRequest, csrfToken) -> secureAuthenticationRequest(securedRequest, csrfToken, "X-XSRF-TOKEN", "XSRF-TOKEN"))
                //gets the SESSION (spring cloud security) token credential and secures the request (adds it to the header and the cookies)
                .zipWith(getCredential(webClientProvider.getInstance(), "spring-cloud"), (securedRequest, sessionToken) -> secureAuthenticationRequest(securedRequest, sessionToken, "Authorization", "SESSION"))
                //does the request and gets the response
                .map(requestBodySpecs -> requestBodySpecs.retrieve())
                //from the response, maps it to a specific DTO. The single() clause is to validate that a body is present.
                .flatMap(clientResponse -> clientResponse.bodyToMono(SecurityCredentialResponseDTO.class).single())
                //checks the authentication and throws a Unauthorizedstatus if its not valid.
                .flatMap(responseDTO -> checkTokenAuthentication(serverWebExchange, responseDTO))
                //if an error is present, then throws it 
                .onErrorResume(e -> {
                    if (e instanceof FWKException.GenericException) {
                        throw (FWKException.GenericException) e;
                    }
                    throw (RuntimeException) e;
                })
                //finally, continues the execution if no exception was thrown.
                .then(webFilterChain.filter(serverWebExchange));

我在这个解决方案中还实现了更多功能（存储CSRF和spring云凭据以避免不必要的调用）。