使用链接响应标头的分页调用中的 Spring Flux

Posted 2023-02-27

【中文标题】使用链接响应标头的分页调用中的 Spring Flux

【英文标题】：Spring Flux from paginated calls using Link response header

【发布时间】：2021-02-25 10:57:51

【问题描述】：

对响应式编程和 Spring webclient 来说相当新，所以我试图了解它，本质上我有一个分页的结果列表，可以从 REST 服务使用，该服务使用带有相对 URI 的链接响应标头进行响应下一页结果（如果存在）。我想将所有页面收集到 Flux 中。默认情况下，生成的 ApiClient 是这样的

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);

那么我是否需要对结果的每一页进行某种递归调用，将每一页 Mono 组合成一个通量？例如。沿着

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));

【参考方案1】：

所以我设法通过将ClientResponse 视为结果页面来实现此功能。

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();