how to configure pooled connection idle timeout in reactor-netty

Question

I am using reactor-netty http client (0.7.X series) with connection pooling and would like to configure pooled connection's idle timeout but don't know where.

More precisely, I need to configure reactor-netty http client connection pool in such a way that it will automatically close connections that did not see any activity within configurable timeout. These connections are open but no bytes were transferred in or out for some (configurable) amount of time.

How can I configure reactory-netty http client to close idle connections preemptively?

Answer 1

I managed to configure WebClient (via underlying TcpClient) to remove idle connections on timeout from connection pool in reactor-netty 0.8.9

My solution is partially based on the official documentation about IdleStateHandler extended with my research on how to properly apply it when creating an instance of HttpClient.

Here is how I did that:

public class IdleCleanupHandler extends ChannelDuplexHandler {
    @Override
    public void userEventTriggered(final ChannelHandlerContext ctx, final Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            final IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.ALL_IDLE) { // or READER_IDLE / WRITER_IDLE
                // close idling channel
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }
}

...

public static WebClient createWebClient(final String baseUrl, final int idleTimeoutSec) {
    final TcpClient tcpClient = TcpClient.create(ConnectionProvider.fixed("fixed-pool"))
        .bootstrap(bootstrap -> BootstrapHandlers.updateConfiguration(bootstrap, "idleTimeoutConfig",
            (connectionObserver, channel) -> {
                channel.pipeline()
                    .addLast("idleStateHandler", new IdleStateHandler(0, 0, idleTimeoutSec))
                    .addLast("idleCleanupHandler", new IdleCleanupHandler());
            }));

    return WebClient.builder()
        .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient)))
        .baseUrl(baseUrl)
        .build();
}

---

IMPORTANT UPDATE:

My further testing has indicated that adding handlers during bootstrap hook distructs the pool and sockets (channels) are not reused by Connection.

The right way to add the handlers is:

public static WebClient createWebClient(final String baseUrl, final int idleTimeoutSec) {
    final TcpClient tcpClient = TcpClient.create(ConnectionProvider.fixed("fixed-pool"))
        .doOnConnected(conn -> {
            final ChannelPipeline pipeline = conn.channel().pipeline();
            if (pipeline.context("idleStateHandler") == null) {
                pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, idleTimeoutSec))
                        .addLast("idleCleanupHandler", new IdleCleanupHandler());
            }
        });

    return WebClient.builder()
        .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient)))
        .baseUrl(baseUrl)
        .build();
}

Note: in reactor-netty 0.9.x there will be a standard way to configure idle timeout for connections in the connection pool, see this commit: https://github.com/reactor/reactor-netty/pull/792

Answer 2

I was able to accomplish this on the 0.7.x branch by adding netty write and read time-out handlers to the channel pipeline. However, on 0.8.x, this approach no longer works.

HttpClient httpClient = HttpClient
    .create((HttpClientOptions.Builder builder) -> builder
    .host(endpointUrl.getHost())
    .port(endpointUrl.getPort())
    .poolResources(PoolResources.fixed(connectionPoolName, maxConnections, timeoutPool))
    .afterChannelInit(channel -> {
        channel.pipeline()
                // The write and read timeouts are serving as generic socket idle state handlers.
                .addFirst("write_timeout", new WriteTimeoutHandler(timeoutIdle, TimeUnit.MILLISECONDS))
                .addFirst("read_timeout", new ReadTimeoutHandler(timeoutIdle, TimeUnit.MILLISECONDS));
    })
    .build());