Netty: обработчик состояния простоя не показывает, если канал свободен - PullRequest
Новая
       6

Netty: обработчик состояния простоя не показывает, если канал свободен

0 голосов
/ 29 апреля 2018

Мои требования:
Я хочу определить, не занят ли канал для чтения в течение некоторого времени, и хочу установить тайм-аут на основании этого. Мой Netty клиент отправляет запрос на 1000 серверов.

Проблема : Мой Netty Client никогда не показывает, что, если какое-то время есть какой-нибудь неактивный канал, даже если я использую некоторые ips, которые всегда будут зависать. Я сомневаюсь, что я не реализую IdleStateHandler правильно. Я попытался уменьшить время ожидания чтения IdleStateHandler, но не повезло. Я потратил часы, чтобы понять это. Любая помощь будет по достоинству оценена.
Итак, ниже мой код:
My Netty Client: 

public void connect(final InetAddress remoteAddress){
        new Bootstrap()
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectionTimeout)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .group(eventLoopGroup)
            .channel(NioSocketChannel.class)
            .handler(httpNettyClientChannelInitializer)
            .connect(remoteAddress, serverPort)
            .addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) {
                        future.cancel(!future.isSuccess());
                    }
                });
    }

Инициализатор My Netty Channel: 

public class HttpNettyClientChannelInitializer extends ChannelInitializer<SocketChannel> {

    private final Provider<HttpNettyClientChannelHandler> handlerProvider;
    private final int timeout;
    private int maxContentLength;

    @Inject
    public HttpNettyClientChannelInitializer(Provider<HttpNettyClientChannelHandler> handlerProvider,
            @Named("readResponseTimeout") int timeout, @Named("maxContentLength") int maxContentLength) {
        this.handlerProvider = handlerProvider;
        this.timeout = timeout;
        this.maxContentLength = maxContentLength;
    }

    @Override
    protected void initChannel(SocketChannel socketChannel) {
        ChannelPipeline pipelineNettyClientChannel = socketChannel.pipeline();
        pipelineNettyClientChannel.addLast("codec", new HttpClientCodec());
        pipelineNettyClientChannel.addLast("idleStateHandler", new IdleStateHandler(timeout,0,0, TimeUnit.MILLISECONDS));
        pipelineNettyClientChannel.addLast("aggregator", new HttpObjectAggregator(maxContentLength));
        //handlerProvider.get() will provide new instance of channel handler
        pipelineNettyClientChannel.addLast("handler", handlerProvider.get());
    }

}

My Netty Channel Handler 

 @ChannelHandler.Sharable
    public class HttpNettyClientChannelHandler extends SimpleChannelInboundHandler<HttpObject> {
      @Override
      public void channelActive(ChannelHandlerContext channelHandlerContext){...}

    @Override    
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, HttpObject httpObject){...}

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
             if (evt instanceof IdleStateEvent) {
                 IdleStateEvent e = (IdleStateEvent) evt;
                 if (e.state() == IdleState.READER_IDLE) {
                     System.out.println("Idle channel");
                     ctx.close();
             }
         }
     }
}
...