Implementando mensajes keep-alive en Netty usando WriteTimeoutHandler

Question

Estoy usando Netty 3.2.7. Intento escribir la funcionalidad en mi cliente de modo que si no se escriben mensajes después de un cierto período de tiempo (digamos, 30 segundos), se envíe un mensaje de "mantener vivo" al servidor.

Después de algunas búsquedas, descubrí que WriteTimeoutHandler debería permitirme hacer esto. Encontré esta explicación aquí: https://issues.jboss.org/browse/NETTY-79.

El ejemplo dado en la documentación Netty es:

public ChannelPipeline getPipeline() { 
    // An example configuration that implements 30-second write timeout: 
    return Channels.pipeline(
     new WriteTimeoutHandler(timer, 30), // timer must be shared. 
     new MyHandler()); 
} 

En mi cliente de prueba, he hecho precisamente esto. En MyHandler, también overrided el método exceptionCaught():

public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) { 
    if (e.getCause() instanceof WriteTimeoutException) { 
     log.info("Client sending keep alive!"); 
     ChannelBuffer keepAlive = ChannelBuffers.buffer(KEEP_ALIVE_MSG_STR.length()); 
     keepAlive.writeBytes(KEEP_ALIVE_MSG_STR.getBytes()); 
     Channels.write(ctx, Channels.future(e.getChannel()), keepAlive); 
    } 
} 

No importa lo que la duración del cliente no escribe nada en el canal, el método exceptionCaught() He anulado nunca es llamado.

En cuanto a la fuente de WriteTimeoutHandler, su aplicación writeRequested() es:

public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) 
     throws Exception { 

    long timeoutMillis = getTimeoutMillis(e); 
    if (timeoutMillis > 0) { 
     // Set timeout only when getTimeoutMillis() returns a positive value. 
     ChannelFuture future = e.getFuture(); 
     final Timeout timeout = timer.newTimeout(
       new WriteTimeoutTask(ctx, future), 
       timeoutMillis, TimeUnit.MILLISECONDS); 

     future.addListener(new TimeoutCanceller(timeout)); 
    } 

    super.writeRequested(ctx, e); 
} 

Aquí, parece que esta implementación dice: "Cuando se solicite una escritura, hacer un nuevo tiempo de espera Cuando la escritura se realiza correctamente. , cancela el tiempo de espera ".

Usando un depurador, parece que esto es lo que está sucediendo. Tan pronto como se completa la escritura, el tiempo de espera se cancela. Este no es el comportamiento que quiero. El comportamiento que deseo es: "Si el cliente no ha escrito ninguna información en el canal durante 30 segundos, ejecute WriteTimeoutException".

Entonces, ¿no es esto para lo que es WriteTimeoutHandler? Así es como lo interpreté a partir de lo que he leído en línea, pero la implementación no parece funcionar de esta manera. ¿Lo estoy usando mal? ¿Debería usar algo más? En nuestra versión Mina del mismo cliente que estoy tratando de reescribir, veo que el método sessionIdle() se reemplaza para lograr el comportamiento que quiero, pero este método no está disponible en Netty.

Answer 1

Sugeriría agregar el IdleStateHandler y luego agregar su implementación personalizada de IdleStateAwareUpstreamHandler que puede reaccionar en el estado inactivo. Esto me funciona muy bien en muchos proyectos diferentes.

Los javadocs lista el siguiente ejemplo, que se puede utilizar como la base de su aplicación:

public class MyPipelineFactory implements ChannelPipelineFactory { 

    private final Timer timer; 
    private final ChannelHandler idleStateHandler; 

    public MyPipelineFactory(Timer timer) { 
     this.timer = timer; 
     this.idleStateHandler = new IdleStateHandler(timer, 60, 30, 0); 
     // timer must be shared. 
    } 

    public ChannelPipeline getPipeline() { 
     return Channels.pipeline(
      idleStateHandler, 
      new MyHandler()); 
    } 
} 

// Handler should handle the IdleStateEvent triggered by IdleStateHandler. 
public class MyHandler extends IdleStateAwareChannelHandler { 

    @Override 
    public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e) { 
     if (e.getState() == IdleState.READER_IDLE) { 
      e.getChannel().close(); 
     } else if (e.getState() == IdleState.WRITER_IDLE) { 
      e.getChannel().write(new PingMessage()); 
     } 
    } 
} 

ServerBootstrap bootstrap = ...; 
Timer timer = new HashedWheelTimer(); 
... 
bootstrap.setPipelineFactory(new MyPipelineFactory(timer)); 
... 

Answer 2

Para Netty 4.0 y posteriores, debe ampliar ChannelDuplexHandler como en el ejemplo de IdleStateHandler documentation:

// An example that sends a ping message when there is no outbound traffic 
// for 30 seconds. The connection is closed when there is no inbound traffic 
// for 60 seconds. 

public class MyChannelInitializer extends ChannelInitializer<Channel> { 
    @Override 
    public void initChannel(Channel channel) { 
     channel.pipeline().addLast("idleStateHandler", new IdleStateHandler(60, 30, 0)); 
     channel.pipeline().addLast("myHandler", new MyHandler()); 
    } 
} 

// Handler should handle the IdleStateEvent triggered by IdleStateHandler. 
public class MyHandler extends ChannelDuplexHandler { 
    @Override 
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { 
     if (evt instanceof IdleStateEvent) { 
      IdleStateEvent e = (IdleStateEvent) evt; 
      if (e.state() == IdleState.READER_IDLE) { 
       ctx.close(); 
      } else if (e.state() == IdleState.WRITER_IDLE) { 
       ctx.writeAndFlush(new PingMessage()); 
      } 
     } 
    } 
}