How to set a timeout in Spring 5 WebFlux WebClient

Question

I'm trying to set timeout on my WebClient, here is the current code :

SslContext sslContext = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();

ClientHttpConnector httpConnector = new ReactorClientHttpConnector(opt -> {
    opt.sslContext(sslContext);
    HttpClientOptions option = HttpClientOptions.builder().build();
    opt.from(option);
});
return WebClient.builder().clientConnector(httpConnector).defaultHeader("Authorization", xxxx)
                .baseUrl(this.opusConfig.getBaseURL()).build();

I need to add timeout and also pooling strategy, I was thinking of something like that :

PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager();
cm.setMaxTotal(this.applicationConfig.getHttpClientMaxPoolSize());
cm.setDefaultMaxPerRoute(this.applicationConfig.getHttpClientMaxPoolSize());
cm.closeIdleConnections(this.applicationConfig.getServerIdleTimeout(), TimeUnit.MILLISECONDS);

RequestConfig requestConfig = RequestConfig.custom().setSocketTimeout(this.applicationConfig.getHttpClientSocketTimeout())
        .setConnectTimeout(this.applicationConfig.getHttpClientConnectTimeout())
        .setConnectionRequestTimeout(this.applicationConfig.getHttpClientRequestTimeout()).build();

CloseableHttpClient httpClient = HttpClients.custom().setDefaultRequestConfig(requestConfig).setConnectionManager(cm).build();

But I can't figure out how to set the httpClient in my webclient

Answer 1

The WebFlux WebClient doesn't use Apache Commons HTTP Client. Although you might be able to implement one solution via custom ClientHttpConnector. The existing ReactorClientHttpConnector is based on the Netty. So, consider to use Netty options to configure the client, e.g.:

ReactorClientHttpConnector connector =
            new ReactorClientHttpConnector(options ->
                    options.option(ChannelOption.SO_TIMEOUT, this.applicationConfig.getHttpClientConnectTimeout()));

or

.onChannelInit(channel -> channel.config().setConnectTimeoutMillis(this.applicationConfig.getHttpClientConnectTimeout()))

UPDATE

We also can use ReadTimeoutHandler:

.onChannelInit(channel -> 
        channel.pipeline()
           .addLast(new ReadTimeoutHandler(this.applicationConfig.getHttpClientConnectTimeout())))

Answer 2

Here's how I did it (thanks to @Artem)

SslContext sslContext = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();

        ClientHttpConnector httpConnector = new ReactorClientHttpConnector(options -> {
            options.sslContext(sslContext);
            options.option(ChannelOption.SO_TIMEOUT, this.applicationConfig.getHttpClientRequestTimeout());
            options.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.applicationConfig.getHttpClientConnectTimeout());
            options.poolResources(PoolResources.fixed("myPool", this.applicationConfig.getHttpClientMaxPoolSize()));
        });

        return WebClient.builder().clientConnector(httpConnector).defaultHeader("Authorization", "xxxx")
                .baseUrl(this.config.getBaseURL()).build();

Answer 3

Kotlin Syntax!!

webClient
.post()
.body(BodyInserters.fromObject(body))
.headers(headersSetter)
.retrieve()
.bodyToMono<SomeClass>()
.timeout(Duration.ofSeconds(30)) /// specify timeout duration
.doOnNext {
    logger.info{ "log something"}
}
.onErrorMap { throwable ->
    logger.error{"do some transformation"}
    throwable
}

Answer 4

You can use overloaded block() method which accepts a timeout on Mono object. Or there is a timeout() method directly available on Mono object.

WebClient webClient = WebClient.builder()
              .baseUrl( "https://test.com" )
              .defaultHeader( HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE )
              .build(); 

webClient.post()
              .uri( "/services/some/uri" )
              .body( Mono.just( YourEntityClassObject ), YourEntityClass.class )
              .retrieve()
              .bodyToMono( String.class )
              .timeout(Duration.ofMillis( 5000 )) // option 1
              .block(Duration.ofMillis( 5000 )); // option 2

Answer 5

ReactorClientHttpConnector API changed in version Spring WebFlux 5.1.

So I do the following (Kotlin syntax, based on @joshiste example):

val tcpClient = TcpClient.create()
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10_000)
    .doOnConnected { connection ->
        connection.addHandlerLast(ReadTimeoutHandler(10))
            .addHandlerLast(WriteTimeoutHandler(10))
    }

val myWebClient = webClientBuilder
    .clientConnector(ReactorClientHttpConnector(HttpClient.from(tcpClient)))
    .baseUrl(myEndPoint)
    .build()

UPDATE 2021

HttpClient.from in deprecated in last version of Reactive Netty. It was copying the configuration of the tcpClient. Now we can configure httpClient directly.

val httpClient = HttpClient.create()
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10_000)
    .doOnConnected { connection ->
        connection.addHandlerLast(ReadTimeoutHandler(10))
            .addHandlerLast(WriteTimeoutHandler(10))
    }

val myWebClient = webClientBuilder
    .clientConnector(ReactorClientHttpConnector(httpClient))
    .baseUrl(myEndPoint)
    .build()

Answer 6

As Spring Webflux was updated, here is a solution that works for Java (based on the answer for Kotlin):

TcpClient timeoutClient = TcpClient.create()
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, SECONDS*1000)
    .doOnConnected(
        c -> c.addHandlerLast(new ReadTimeoutHandler(SECONDS))
              .addHandlerLast(new WriteTimeoutHandler(SECONDS)));
return webClientBuilder.baseUrl(YOUR_URL)
       .clientConnector(new ReactorClientHttpConnector(HttpClient.from(timeoutClient)))
       .build();

UPDATE 2021

since HttpClient.from(TcpClient) is deprectaed now it's even easier:

return WebClient.builder()
                .baseUrl(YOUR_URL)
                .clientConnector(new ReactorClientHttpConnector(HttpClient.create()
                                                                          .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, SECONDS * 1000)
                                                                          .doOnConnected(c -> c.addHandlerLast(new ReadTimeoutHandler(SECONDS))
                                                                                               .addHandlerLast(new WriteTimeoutHandler(SECONDS)))))
                .build();

Answer 7

Rather than creating your own WebClient.Builder, you can provide a custom ReactorNettyHttpClientMapper which is going to be applied to the default WebClient.Builder:

@Configuration
class MyAppConfiguration {

    @Bean
    fun reactorNettyHttpClientMapper(): ReactorNettyHttpClientMapper {
        return ReactorNettyHttpClientMapper { httpClient ->
            httpClient.tcpConfiguration { tcpClient ->
                tcpClient.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30_000)
                    .doOnConnected { connection ->
                        connection.addHandlerLast(ReadTimeoutHandler(60))
                            .addHandlerLast(WriteTimeoutHandler(60))
                    }
            }
        }
    }
}

Answer 8

To set the read and connect timeout I use the method below, because the SO_TIMEOUT option is not available for channels using NIO (and giving the warning Unknown channel option 'SO_TIMEOUT' for channel '[id: 0xa716fcb2]')

ReactorClientHttpConnector connector = new ReactorClientHttpConnector(
          options -> options.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 2000)
                            .compression(true)
                            .afterNettyContextInit(ctx -> {
                                ctx.addHandlerLast(new ReadTimeoutHandler(5000, TimeUnit.MILLISECONDS));
                            }));
return WebClient.builder()
                .clientConnector(connector)
                .build();

Answer 9

Based on the above comment if you want to add a Socket Timeout just add it as another option in the same timeoutClient.

TcpClient timeoutClient = TcpClient.create()
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, SECONDS*10) //Connect Timeout
    .option(ChannelOption.SO_TIMEOUT,1000) // Socket Timeout
    .doOnConnected(
        c -> c.addHandlerLast(new ReadTimeoutHandler(SECONDS))
              .addHandlerLast(new WriteTimeoutHandler(SECONDS)));
return webClientBuilder.baseUrl(YOUR_URL)
       .clientConnector(new ReactorClientHttpConnector(HttpClient.from(timeoutClient)))
       .build();

Answer 10

With Spring Webflux 5.1.8 I ran into problems yielding the error messages below using the answer from mcoolive when executing multiple subsequent tests that uses the WebClient.

Force-closing a channel whose registration task was not accepted by an event loop
Failed to submit a listener notification task. Event loop shut down?

Adding a connection provider and loop resources solved my problem:

final ConnectionProvider theTcpClientPool = ConnectionProvider.elastic("tcp-client-pool");
final LoopResources theTcpClientLoopResources = LoopResources.create("tcp-client-loop");

final TcpClient theTcpClient = TcpClient
    .create(theTcpClientPool)
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
    .runOn(theTcpClientLoopResources)
    .doOnConnected(theConnection -> {
        theConnection.addHandlerLast(new ReadTimeoutHandler(mTimeoutInMillisec, TimeUnit.MILLISECONDS));
        theConnection.addHandlerLast(new WriteTimeoutHandler(mTimeoutInMillisec, TimeUnit.MILLISECONDS));
    });

WebClient theWebClient = WebClient.builder()
    .baseUrl(mVfwsServerBaseUrl)
    .clientConnector(new ReactorClientHttpConnector(HttpClient.from(theTcpClient)))
    .build();