как настроить время простоя пула подключения в реактор-нетты

Question

Я использую http-клиент реактор-нетти (серия 0.7.X) с пулом соединений и хотел бы настроить время ожидания простоя пула, но не знаю, где.

Точнее, мне нужно настроить реактор-нетти пул соединений клиента http таким образом, чтобы он автоматически закрывал соединения, которые не видели никакой активности в течение настраиваемого тайм-аута.Эти соединения открыты, но в течение некоторого (настраиваемого) промежутка времени байты не передавались и не выводились.

Как я могу настроить http-клиент реактора-нетто для упреждающего закрытия незанятых соединений?

Answer 1

Мне удалось настроить WebClient (через базовый TcpClient) для удаления незанятых соединений по таймауту из пула соединений в реактор-нетто 0,8,9

Мое решение частично основанов официальной документации о IdleStateHandler , дополненной моим исследованием того, как правильно применять его при создании экземпляра HttpClient.

Вот как я это сделал:

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

---

ВАЖНОЕ ОБНОВЛЕНИЕ:

Мое дальнейшее тестирование показало, что добавление обработчиков во время bootstrap перебивает пул, и сокеты (каналы) не используются Connection повторно.

Правильный способ добавить обработчики:

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

Примечание: в reactor-netty 0.9.x будет стандартный способ настройки времени простоя для соединений в пуле соединений,увидеть этот коммит: https://github.com/reactor/reactor-netty/pull/792

Answer 2

Мне удалось сделать это на ветке 0.7.x, добавив обработчики времени записи и чтения нетто в конвейер канала.Однако на 0.8.x этот подход больше не работает.

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