Cliente Web

Spring WebFlux contiene un cliente para realizar solicitudes HTTP. WebClient tiene una API fluida y funcional basada en Reactor que le permite componer lógica asincrónica de forma declarativa sin la necesidad de lidiar con subprocesos o concurrencia. Es completamente sin bloqueo, admite transmisión y se basa en los mismos códecs utilizados para codificar y decodificar el contenido de las solicitudes y respuestas en el lado del servidor.

WebClient requiere un Biblioteca de cliente HTTP para ejecutar solicitudes. Hay soporte integrado para:

• Reactor Netty

• Jetty Reactive HttpClient

• Componentes Http de Apache

• Se pueden conectar otros clientes a través de ClientHttpConnector.

Configuración

El La forma más sencilla de crear WebClient es utilizar uno de los métodos estáticos de fábrica:

• WebClient.create()

• WebClient.create(String baseUrl)

También puedes usar WebClient.builder() con parámetros adicionales:

• uriBuilderFactory: UriBuilderFactory configurado para ser utilizado como URL base.< /p>

• defaultUriVariables: valores predeterminados que se utilizarán al expandir las plantillas de URI.

• defaultHeader: encabezados para cada solicitud.

• defaultCookie: cookies para cada solicitud.

• defaultRequest: Consumidor para configurar cada solicitud.

• filter: Filtro de cliente para cada solicitud.

• exchangeStrategies: Configuración para leer/escribir mensajes HTTP.

• clientConnector: configuración de la biblioteca HTTP del cliente.

Por ejemplo

WebClient client = WebClient.builder()
.codecs(configurer -> ... )
.build();

val webClient = WebClient.builder()
.codecs { configurer -> ... }
.build()

Una vez creado, el WebClient es inmutable. Sin embargo, puedes clonarlo y crear una copia modificada de la siguiente manera:

WebClient client1 = WebClient.builder()
.filter(filterA).filter(filterB).build();
WebClient client2 = client1.mutate()
.filter(filterC).filter(filterD).build();
// cliente1 tiene filterA, filterB
// cliente2 tiene filterA, filterB, filterC, filterD

val client1 = WebClient.builder()
.filter(filterA).filter(filterB).build()
val client2 = client1.mutate()
.filter(filterC).filter(filterD).build()
// cliente1 tiene filterA, filterB
// cliente2 tiene filterA, filterB, filterC, filterD

MaxInMemorySize

Los códecs tienen limitaciones en el almacenamiento intermedio de datos en la memoria para evitar problemas de memoria de la aplicación. Por defecto están configurados en 256 KB. Si esto no es suficiente, se recibirá el siguiente error:

org.springframework.core.io.buffer.DataBufferLimitException: se superó el límite de bytes máximos en el búfer

Para cambiar el límite predeterminado para códecs, haga lo siguiente:

WebClient webClient = WebClient.builder()
.codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(2 * 1024 * 1024))
.build();

val webClient = WebClient.builder()
.codecs { configurer -> configurer.defaultCodecs().maxInMemorySize(2 * 1024 * 1024) }
.build()

Reactor Netty

Para configurar los parámetros de Reactor Netty, proporcione una configuración preconfigurada. HttpClient:

HttpClient httpClient = HttpClient.create().secure(sslSpec -> ...);
WebClient webClient = WebClient.builder()
.clientConnector(new ReactorClientHttpConnector(httpClient))
.build();

val httpClient = HttpClient.create().secure { ... }
val webClient = WebClient.builder()
.clientConnector(ReactorClientHttpConnector(httpClient))
.build()

Recursos

De forma predeterminada, HttpClient participa en el uso de los recursos globales de Reactor Netty almacenados en reactor.netty.http.HttpResources, incluidos los subprocesos y el grupo de bucles de eventos. conexiones. Se recomienda este modo porque, a efectos de simultaneidad en los bucles de espera de eventos, es preferible utilizar recursos fijos compartidos. En este modo, los recursos globales permanecen activos hasta que finaliza el proceso.

Si el servidor está sincronizado con el proceso, normalmente no hay necesidad de finalizar el proceso explícitamente. Sin embargo, si el servidor se puede iniciar o detener durante el proceso (como es el caso de una aplicación Spring MVC implementada como un archivo WAR), entonces puede declarar un bean administrado Spring de tipo ReactorResourceFactory con el parámetro globalResources=true (predeterminado) para que se garantice que el uso de los recursos globales de Reactor Netty finalice cuando ApplicationContext se cierre desde Spring, como se muestra en el siguiente ejemplo:

@Bean
public ReactorResourceFactory reactorResourceFactory() {
return new ReactorResourceFactory();
}

@Bean
fun reactorResourceFactory() = ReactorResourceFactory()

También puede evitar el uso de recursos globales de Reactor Netty. Sin embargo, en este modo, usted es responsable de garantizar que todas las instancias del cliente y servidor Reactor Netty compartan recursos, como se muestra en el siguiente ejemplo:

@Bean
public ReactorResourceFactory resourceFactory() {
ReactorResourceFactory factory = new ReactorResourceFactory();
factory.setUseGlobalResources(false); 
return factory;
}
@Bean
public WebClient webClient() {
Function<HttpClient, HttpClient> mapper = client -> {
// Configuración adicional...
};
ClientHttpConnector connector =
    new ReactorClientHttpConnector(resourceFactory(), mapper); 
return WebClient.builder().clientConnector(connector).build(); 
}

@Bean
fun resourceFactory() = ReactorResourceFactory().apply {
isUseGlobalResources = false 
}
@Bean
fun webClient(): WebClient {
val mapper: (HttpClient) -> HttpClient = {
// Configuración adicional...
}
val connector = ReactorClientHttpConnector(resourceFactory(), mapper) 
return WebClient.builder().clientConnector(connector).build() 
}

Tiempo de espera

Configuración de valores de tiempo de espera de conexión:

import io.netty.channel.ChannelOption;
HttpClient httpClient = HttpClient.create()
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000);
WebClient webClient = WebClient.builder()
.clientConnector(new ReactorClientHttpConnector(httpClient))
.build();

import io.netty.channel.ChannelOption
val httpClient = HttpClient.create()
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000);
val webClient = WebClient.builder()
.clientConnector(new ReactorClientHttpConnector(httpClient))
.build();

Configuración del valor de tiempo de espera de lectura y escritura:

import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
HttpClient httpClient = HttpClient.create()
.doOnConnected(conn -> conn
        .addHandlerLast(new ReadTimeoutHandler(10))
        .addHandlerLast(new WriteTimeoutHandler(10)));
// Crear un cliente web...

 
import io.netty.handler.timeout.ReadTimeoutHandler
import io.netty.handler.timeout.WriteTimeoutHandler
val httpClient = HttpClient.create()
.doOnConnected { conn -> conn
        .addHandlerLast(new ReadTimeoutHandler(10))
        .addHandlerLast(new WriteTimeoutHandler(10))
}// Crear un cliente web...

Configurar el tiempo de espera de respuesta para una solicitud específica:

HttpClient httpClient = HttpClient.create()
.responseTimeout(Duration.ofSeconds(2));
// Crear un cliente web...

 
val httpClient = HttpClient.create()
.responseTimeout(Duration.ofSeconds(2));
// Crear un cliente web...

El siguiente ejemplo muestra cómo configurar el HttpClient de Jetty:

WebClient.create().get()
.uri("https://example.org/path")
.httpRequest(httpRequest -> {
    HttpClientRequest reactorRequest = httpRequest.getNativeRequest();
    reactorRequest.responseTimeout(Duration.ofSeconds(2));
})
.retrieve()
.bodyToMono(String.class);

WebClient.create().get()
.uri("https://example.org/path")
.httpRequest { httpRequest: ClientHttpRequest ->
    val reactorRequest = httpRequest.getNativeRequest<HttpClientRequest>()
    reactorRequest.responseTimeout(Duration.ofSeconds(2))
}
.retrieve()
.bodyToMono(String::class.java)

Jetty

El siguiente ejemplo muestra cómo configurar los ajustes de HttpClient desde Jetty:

HttpClient httpClient = new HttpClient();
httpClient.setCookieStore(...);
WebClient webClient = WebClient.builder()
.clientConnector(new JettyClientHttpConnector(httpClient))
.build();

 val httpClient = HttpClient()
httpClient.cookieStore = ...
val webClient = WebClient.builder()
.clientConnector(new JettyClientHttpConnector(httpClient))
.build();

Por de forma predeterminada, HttpClient crea sus propios recursos (Ejecutor, ByteBufferPool, Scheduler) que permanecen activos hasta que se completa el proceso. ejecución o llamada a la función stop().

Es posible compartir recursos entre múltiples instancias del cliente (y servidor) Jetty y garantizar que el uso de recursos finalice cuando ApplicationContext se cierra desde Spring, declarando un bean administrado por Spring de tipo JettyResourceFactory, como se muestra en el siguiente ejemplo:

@Bean
public JettyResourceFactory resourceFactory() {
return new JettyResourceFactory();
}
@Bean
public WebClient webClient() {
HttpClient httpClient = new HttpClient();
// Configuración adicional...
ClientHttpConnector connector =
    new JettyClientHttpConnector(httpClient, resourceFactory()); 
return WebClient.builder().clientConnector(connector).build(); 
}

@Bean
fun resourceFactory() = JettyResourceFactory()
@Bean
fun webClient(): WebClient {
val httpClient = HttpClient()
// Configuración adicional...
val connector = JettyClientHttpConnector(httpClient, resourceFactory()) 
return WebClient.builder().clientConnector(connector).build() 
}

HttpComponents

El siguiente ejemplo muestra cómo configurar los parámetros HttpClient de Apache HttpComponents:

HttpAsyncClientBuilder clientBuilder = HttpAsyncClients.custom();
clientBuilder.setDefaultRequestConfig(...);
CloseableHttpAsyncClient client = clientBuilder.build();
ClientHttpConnector connector = new HttpComponentsClientHttpConnector(client);
WebClient webClient = WebClient.builder().clientConnector(connector).build();

val client = HttpAsyncClients.custom().apply {
setDefaultRequestConfig(...)
}.build()
val connector = HttpComponentsClientHttpConnector(client)
val webClient = WebClient.builder().clientConnector(connector).build()

retrieve()

El método retrieve() se puede utilizar para declarar cómo recuperar la respuesta. Por ejemplo:

WebClient client = WebClient.create("https://example.org");
Mono<ResponseEntity<Person>> result = client.get()
.uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
.retrieve()
.toEntity(Person.class);

val client = WebClient.create("https://example.org")
val result = client.get()
.uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
.retrieve()
.toEntity<Person>().awaitSingle()

O obtenemos solo el cuerpo:

WebClient client = WebClient.create("https://example.org");
Mono<Person> result = client.get()
.uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
.retrieve()
.bodyToMono(Person.class);

val client = WebClient.create("https://example.org")
val result = client.get()
.uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
.retrieve()
.awaitBody<Person>()

Obtener una secuencia de objetos decodificados:

Flux<Quote> result = client.get()
.uri("/quotes").accept(MediaType.TEXT_EVENT_STREAM)
.retrieve()
.bodyToFlux(Quote.class);

val result = client.get()
.uri("/quotes").accept(MediaType.TEXT_EVENT_STREAM)
.retrieve()
.bodyToFlow<Quote>()

De forma predeterminada, las respuestas 4xx o 5xx dan como resultado una WebClientResponseException, incluidas subclases para ciertos códigos de estado HTTP . Para configurar cómo se manejan los mensajes de error, use los controladores onStatus de la siguiente manera:

Mono<Person> result = client.get()
.uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
.retrieve()
.onStatus(HttpStatus::is4xxClientError, response -> ...)
.onStatus(HttpStatus::is5xxServerError, response -> ...)
.bodyToMono(Person.class);

val result = client.get()
.uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
.retrieve()
.onStatus(HttpStatus::is4xxClientError) { ... }
.onStatus(HttpStatus::is5xxServerError) { ... }
.awaitBody<Person>()

Exchange

Métodos exchangeToMono() y exchangeToFlux() (o awaitExchange { } y exchangeToFlow { } en Kotlin) útil para casos más complejos que requieren más control, como decodificar una respuesta de manera diferente según el estado de la respuesta:

Mono<Person> entityMono = client.get()
.uri("/persons/1")
.accept(MediaType.APPLICATION_JSON)
.exchangeToMono(response -> {
    if (response.statusCode().equals(HttpStatus.OK)) {
        return response.bodyToMono(Person.class);
    }
    else {
        // Solucionar el error
        return response.createException().flatMap(Mono::error);
    }
});

val entity = client.get()
.uri("/persons/1")
.accept(MediaType.APPLICATION_JSON)
.awaitExchange {
if (response.statusCode() == HttpStatus.OK) {
     return response.awaitBody<Person>()
}
else {
     throw response.createExceptionAndAwait()
}
}}

Cuando se utiliza el código anterior, después de completar el Mono o devuelto Flux , el cuerpo de la respuesta se verifica y, si no se usa, se libera para evitar pérdidas de memoria y conexión. Por lo tanto, la respuesta no se puede decodificar más adelante. La función proporcionada debe determinar cómo decodificar la respuesta si es necesario.

Cuerpo de la solicitud

El cuerpo de la solicitud se puede codificar desde cualquier tipo asincrónico manejado por ReactiveAdapterRegistry , como Mono o Deferred de las corrutinas de Kotlin, como se muestra en el siguiente ejemplo:

Mono<Person> personMono = ... ;
Mono<Void> result = client.post()
.uri("/persons/{id}", id)
.contentType(MediaType.APPLICATION_JSON)
.body(personMono, Person.class)
.retrieve()
.bodyToMono(Void.class);
    

val personDeferred: Deferred<Person> = ...
client.post()
.uri("/persons/{id}", id)
.contentType(MediaType.APPLICATION_JSON)
.body<Person>(personDeferred)
.retrieve()
.awaitBody<Unit>()

Flux<Person> personFlux = ... ;
Mono<Void> result = client.post()
.uri("/persons/{id}", id)
.contentType(MediaType.APPLICATION_STREAM_JSON)
.body(personFlux, Person.class)
.retrieve()
.bodyToMono(Void.class);

val people: Flow<Person> = ...
client.post()
.uri("/persons/{id}", id)
.contentType(MediaType.APPLICATION_JSON)
.body(people)
.retrieve()
.awaitBody<Unit>()

Person person = ... ;
Mono<Void> result = client.post()
.uri("/persons/{id}", id)
.contentType(MediaType.APPLICATION_JSON)
.bodyValue(person)
.retrieve()
.bodyToMono(Void.class);

val person: Person = ...
client.post()
.uri("/persons/{id}", id)
.contentType(MediaType.APPLICATION_JSON)
.bodyValue(person)
.retrieve()
.awaitBody<Unit>()

MultiValueMap<String, String> formData = ... ;
Mono<Void> result = client.post()
.uri("/path", id)
.bodyValue(formData)
.retrieve()
.bodyToMono(Void.class);

val formData: MultiValueMap<String, String> = ...
client.post()
.uri("/path", id)
.bodyValue(formData)
.retrieve()
.awaitBody<Unit>()

import static org.springframework.web.reactive.function.BodyInserters.*;
Mono<Void> result = client.post()
.uri("/path", id)
.body(fromFormData("k1", "v1").with("k2", "v2"))
.retrieve()
.bodyToMono(Void.class);

import org.springframework.web.reactive.function.BodyInserters.*
client.post()
.uri("/path", id)
.body(fromFormData("k1", "v1").with("k2", "v2"))
.retrieve()
.awaitBody<Unit>()

MultipartBodyBuilder builder = new MultipartBodyBuilder();
builder.part("fieldPart", "fieldValue");
builder.part("filePart1", new FileSystemResource("...logo.png"));
builder.part("jsonPart", new Person("Jason"));
builder.part("myPart", part); // Part from a server request
MultiValueMap<String, HttpEntity<?>> parts = builder.build();

val builder = MultipartBodyBuilder().apply {
part("fieldPart", "fieldValue")
part("filePart1", new FileSystemResource("...logo.png"))
part("jsonPart", new Person("Jason"))
part("myPart", part) // Part from a server request
}
val parts = builder.build()

MultipartBodyBuilder builder = ...;
Mono<Void> result = client.post()
.uri("/path", id)
.body(builder.build())
.retrieve()
.bodyToMono(Void.class);

val builder: MultipartBodyBuilder = ...
client.post()
.uri("/path", id)
.body(builder.build())
.retrieve()
.awaitBody<Unit>()

import static org.springframework.web.reactive.function.BodyInserters.*;
Mono<Void> result = client.post()
.uri("/path", id)
.body(fromMultipartData("fieldPart", "value").with("filePart", resource))
.retrieve()
.bodyToMono(Void.class);

import org.springframework.web.reactive.function.BodyInserters.*
client.post()
.uri("/path", id)
.body(fromMultipartData("fieldPart", "value").with("filePart", resource))
.retrieve()
.awaitBody<Unit>()

WebClient client = WebClient.builder()
.filter((request, next) -> {
    ClientRequest filtered = ClientRequest.from(request)
            .header("foo", "bar")
            .build();
    return next.exchange(filtered);
})
.build();

val client = WebClient.builder()
.filter { request, next ->
    val filtered = ClientRequest.from(request)
            .header("foo", "bar")
            .build()
    next.exchange(filtered)
}
.build()

import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
WebClient client = WebClient.builder()
.filter(basicAuthentication("user", "password"))
.build();

import org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication
val client = WebClient.builder()
.filter(basicAuthentication("user", "password"))
.build()

import static org.springframework.web.reactive.function.client.ExchangeFilterFunctions.basicAuthentication;
WebClient client = webClient.mutate()
.filters(filterList -> {
    filterList.add(0, basicAuthentication("user", "password"));
})
.build();

val client = webClient.mutate()
.filters { it.add(0, basicAuthentication("user", "password")) }
.build()

 public ExchangeFilterFunction renewTokenFilter() {
return (request, next) -> next.exchange(request).flatMap(response -> {
if (response.statusCode().value() == HttpStatus.UNAUTHORIZED.value()) {
    return response.releaseBody()
            .then(renewToken())
            .flatMap(token -> {
                ClientRequest newRequest = ClientRequest.from(request).build();
                return next.exchange(newRequest);
            });
} else {
    return Mono.just(response);
}
});
}

fun renewTokenFilter(): ExchangeFilterFunction? {
return ExchangeFilterFunction { request: ClientRequest?, next: ExchangeFunction ->
next.exchange(request!!).flatMap { response: ClientResponse ->
    if (response.statusCode().value() == HttpStatus.UNAUTHORIZED.value()) {
        return@flatMap response.releaseBody()
                .then(renewToken())
                .flatMap { token: String? ->
                    val newRequest = ClientRequest.from(request).build()
                    next.exchange(newRequest)
                }
    } else {
        return@flatMap Mono.just(response)
    }
}
}
}

WebClient client = WebClient.builder()
.filter((request, next) -> {
    Optional<Object> usr = request.attribute("myAttribute");
    // ...
})
.build();
client.get().uri("https://example.org/")
.attribute("myAttribute", "...")
.retrieve()
.bodyToMono(Void.class);
}

val client = WebClient.builder()
.filter { request, _ ->
    val usr = request.attributes()["myAttribute"];
    // ...
}
.build()
client.get().uri("https://example.org/")
    .attribute("myAttribute", "...")
    .retrieve()
    .awaitBody<Unit>())

WebClient client = WebClient.builder()
.filter((request, next) ->
        Mono.deferContextual(contextView -> {
            String value = contextView.get("foo");
            // ...
        }))
.build();
client.get().uri("https://example.org/")
.retrieve()
.bodyToMono(String.class)
.flatMap(body -> {
        // execute a nested query (the context is propagated automatically)...
})
.contextWrite(context -> context.put("foo", ...));

Person person = client.get().uri("/person/{id}", i).retrieve()
.bodyToMono(Person.class)
.block();
List<Person> persons = client.get().uri("/persons").retrieve()
.bodyToFlux(Person.class)
.collectList()
.block();;

val persona = runBlocking {
client.get().uri("/person/{id}", i).retrieve( )
    .awaitBody<Persona>()
}
val personas = ejecutarBloqueo {
cliente.get().uri("/persons").retrieve()
    .bodyToFlow<Persona>()
    .Listar()
}

Mono<Persona> personMono = client.get().uri("/person/{id}", personId)
.retrieve().bodyToMono(Persona.clase);
Mono<Lista<Hobby>> aficionesMono = client.get().uri("/persona/{id}/aficiones", personId)
.retrieve().bodyToFlux(Hobby.class).collectList();
Mapa<Cadena, Objeto> datos = Mono.zip(personaMono, aficionesMono, (persona, aficiones) -> {
    Mapa<Cadena, Cadena> mapa = nuevo LinkedHashMap<>();
    map.put("persona", persona);
    map.put("aficiones", aficiones);
    regresar mapa;
})
.bloque();

val datos = runBlocking {
val personaDeferred = async {
    cliente.get().uri("/person/{id}", personId)
            .retrieve().awaitBody<Persona>()
}
val pasatiemposDeferidos = asíncrono {
    cliente.get().uri("/person/{id}/hobbies", personId)
            .retrieve().bodyToFlow<Hobby>().toList()
}
mapOf("persona" a personDeferred.await(), "hobbies" a hobbiesDeferred.await())
}