Spring Flux来自使用Link响应头的分页调用

Question

对于反应式编程和SpringWebClient来说，这是一个相当新的概念，所以我正在尝试了解它，本质上，我有一个分页的结果列表，可以从一个REST服务中使用，该服务使用一个链接响应头进行响应，该响应头带有一个与下一页结果（如果存在）相关的URI。我想把所有的页面都收集到一个通量中。默认情况下，生成的ApicClient具有如下内容

public <T> Flux<T> invokeFluxAPI(String path, HttpMethod method, Map<String, Object> pathParams,
        MultiValueMap<String, String> queryParams, Object body, HttpHeaders headerParams,
        MultiValueMap<String, String> cookieParams, MultiValueMap<String, Object> formParams,
        List<MediaType> accept, MediaType contentType, String[] authNames, ParameterizedTypeReference<T> returnType)
        throws RestClientException {
    final WebClient.RequestBodySpec requestBuilder = prepareRequest(path, method, pathParams, queryParams, body,
            headerParams, cookieParams, formParams, accept, contentType, authNames);
    return requestBuilder.retrieve().bodyToFlux(returnType);
}

那么，我是否需要对结果的每一页进行某种递归调用，将每一页合并成一个流量？沿着……的路线

public <T> Flux<T> invokeFluxAPI(String path, HttpMethod method, Map<String, Object> pathParams,
        MultiValueMap<String, String> queryParams, Object body, HttpHeaders headerParams,
        MultiValueMap<String, String> cookieParams, MultiValueMap<String, Object> formParams,
        List<MediaType> accept, MediaType contentType, String[] authNames, ParameterizedTypeReference<T> returnType)
        throws RestClientException {
    final WebClient.RequestBodySpec requestBuilder = prepareRequest(path, method, pathParams, queryParams, body,
            headerParams, cookieParams, formParams, accept, contentType, authNames);
    return requestBuilder.exchange()
            .doOnSuccess(response -> {
                List<String> links = response.headers().header("Link");
                if (!links.isEmpty()) {
                    String nextPageUri = links.get(0);
                    // Get next page using the URI
                }
            })
            .flatMapMany(response -> response.bodyToFlux(returnType));
}

Answer 1

所以我设法让这个工作通过处理客户端响应作为一个结果页eg.

public <T> Flux<T> invokeFluxAPI(String path, HttpMethod method, Map<String, Object> pathParams,
        MultiValueMap<String, String> queryParams, Object body, HttpHeaders headerParams,
        MultiValueMap<String, String> cookieParams, MultiValueMap<String, Object> formParams,
        List<MediaType> accept, MediaType contentType, String[] authNames, ParameterizedTypeReference<T> returnType)
        throws RestClientException {
    return fetchResultsPage(path, method, pathParams, queryParams, body, headerParams, cookieParams, formParams, accept, contentType, authNames).expand(response -> {
        List<String> links = response.headers().header("Link");
        if (links.isEmpty()) {
            return Mono.empty();
        }

        Matcher m = "(?<=\\<)([^\\]]+)(?=\\>)".matcher(links.get(0));

        if (!m.find())) {
            return Mono.empty();
        }

        String nextPageUri = m.group(1);

        MultiValueMap<String, String> qParams = UriComponentsBuilder.fromUriString(nextPageUri).build().getQueryParams();

        queryParams.putAll(qParams);
        queryParams.replace("limit", qParams.get("limit"));

        return fetchResultsPage(path, method, pathParams, queryParams, body, headerParams, cookieParams, formParams, accept, contentType, authNames);
    }).flatMap(response -> response.bodyToFlux(returnType);
}

private Mono<ClientResponse> fetchResultsPage(String path, HttpMethod method, Map<String, Object> pathParams,
        MultiValueMap<String, String> queryParams, Object body, HttpHeaders headerParams,
        MultiValueMap<String, String> cookieParams, MultiValueMap<String, Object> formParams,
        List<MediaType> accept, MediaType contentType, String[] authNames) {
    final WebClient.RequestBodySpec requestBuilder = prepareRequest(path, method, pathParams, queryParams, body,
            headerParams, cookieParams, formParams, accept, contentType, authNames);
    return requestBuilder.exchange();
}