如何使用 Mono 的标头来丰富 ReactiveElasticsearchClient？

Question

我将使用从 ReactiveSecurityContextHolder 检索到的 JWT（作为授权标头）来丰富 Spring Data Elasticsearch -

ReactiveElasticsearchClient

请求。

ReactiveElasticsearchClient

的已弃用版本位于

org.springframework.data.elasticsearch.client.erhlc

包基于Spring的Webclient。所以像下面这样是可能的。

enter code here
@Bean
public ReactiveElasticsearchClient deprecatedReactiveElasticsearchClient() {
    final var clientConfiguration = ClientConfiguration.builder()
        .connectedTo(elasticProperties.getUris().toArray(String[]::new))
        .withClientConfigurer(
            ReactiveRestClients.WebClientConfigurationCallback.from(
                webClient -> webClient.mutate().filter(jwtExchangeFilter()).build())
         )
         .build();
    return ReactiveRestClients.create(clientConfiguration);
}


private ExchangeFilterFunction jwtExchangeFilter() {
    return ExchangeFilterFunction.ofRequestProcessor(request -> retrieveJwt()
        .map(jwt -> ClientRequest.from(request).header(HttpHeaders.AUTHORIZATION, "Bearer " + jwt).build())
        .switchIfEmpty(Mono.defer(() -> Mono.just(ClientRequest.from(request).build()))));
}

private Mono<String> retrieveJwt() {
    return ReactiveSecurityContextHolder.getContext()
        .map(SecurityContext::getAuthentication)
        .map(Authentication::getPrincipal)
        .filter(principle -> principle instanceof Jwt)
        .map(Jwt.class::cast)
        .map(Jwt::getTokenValue);
}
enter code here

但是，放置在

ReactiveElasticsearchClient

包中的

org.springframework.data.elasticsearch.client.elc

的未弃用版本是基于 Apache 的

HttpAsyncClient

，我不知道如何提供具有由

Mono

提供的值的 auth 标头。有一个

ElasticsearchClients.ElasticsearchHttpClientConfigurationCallback.from

方法可以提供

HttpAsyncClientBuilder

。构建器具有添加

org.apache.http.HttpRequestInterceptor

等功能，但在我看来，当通过

Mono

提供值时，这些拦截器在反应式环境中不可用。

最后，

ClientConfiguration

（可在已弃用和未弃用的客户端中使用）提供了接受

withHeader

的

Supplier

方法。但是我想没有办法用

Mono

以非阻塞方式提供

Supplier

的值。或者有什么办法吗？

我尝试使用客户端配置回调和标头供应商配置未弃用的

ReactiveElasticsearchClient

，但无法以非阻塞方式提供由

Mono

（此处为

ReactiveSecurityContextHolder.getContext()

）提供的标头值。我希望这在反应式客户端中是可能的。不是吗？

https://docs.spring.io/spring-data/elasticsearch/docs/current/reference/html/#elasticsearch.clients.configuration

Answer 1

自从切换到新的 Elasticsearch 客户端后，底层的 http 客户端就是 Elasticsearch Transport 提供的客户端，正如您正确提到的那样，这是一种非响应式客户端。

ReactiveElasticsearchClient

使用传输的异步方法并将其放入

Mono.fromFuture()

调用（或 Flux）中。

因此目前无法从反应式代码流添加标头。

在最新版本的 Elasticsearch 客户端 (8.9) 中，对传输进行了一些重写，以便客户端库的用户能够提供自己的传输 (https://github.com/elastic/elasticsearch-java/ issues/550 由 PR https://github.com/elastic/elasticsearch-java/pull/584 关闭），但我还没有时间检查这是否也允许插入 WebClient 以及这将如何然后使用 Spring Data Elasticsearch 中的反应式代码；还有一张与此相关的票证https://github.com/spring-projects/spring-data-elasticsearch/issues/2604。