Implementazione di messaggi keep-alive in Netty utilizzando WriteTimeoutHandler

Question

Sto utilizzando Netty 3.2.7. Sto cercando di scrivere funzionalità nel mio client in modo tale che se nessun messaggio viene scritto dopo un certo periodo di tempo (diciamo 30 secondi), un messaggio "keep-alive" viene inviato al server.

Dopo alcuni scavi, ho scoperto che WriteTimeoutHandler dovrebbe consentirmi di farlo. Ho trovato questa spiegazione qui: https://issues.jboss.org/browse/NETTY-79.

L'esempio riportato nella documentazione di Netty è:

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

Nel mio client di prova, mi hanno fatto proprio questo. In MyHandler, ho anche overrided il metodo 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); 
    } 
} 

Non importa quale sia la durata il cliente non scrive nulla al canale, il metodo exceptionCaught() ho sovrascritto non viene mai chiamato.

Guardando la fonte della WriteTimeoutHandler, la sua attuazione writeRequested() è:

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

Qui, sembra che questa implementazione dice: "Quando si richiede una scrittura, fare un nuovo timeout Quando succede al di scrittura. , annulla il timeout. "

Utilizzando un debugger, sembra che questo sia ciò che sta accadendo. Non appena la scrittura termina, il timeout viene cancellato. Questo non è il comportamento che voglio. Il comportamento che voglio è: "Se il client non ha scritto alcuna informazione sul canale per 30 secondi, lancia una WriteTimeoutException."

Quindi, questo non è ciò per cui WriteTimeoutHandler? È così che l'ho interpretato da ciò che ho letto online, ma l'implementazione non sembra funzionare in questo modo. Sto usando male? Dovrei usare qualcos'altro? Nella nostra versione Mina dello stesso client che sto cercando di riscrivere, vedo che il metodo sessionIdle() viene sovrascritto per ottenere il comportamento che voglio, ma questo metodo non è disponibile in Netty.

Answer 1

Vorrei suggerire di aggiungere IdleStateHandler e quindi aggiungere l'implementazione personalizzata di IdleStateAwareUpstreamHandler che può reagire allo stato di inattività. Questo funziona molto bene per me su molti progetti diversi. Lista

I javadocs il seguente esempio, che si potrebbe usare come base della vostra implementazione:

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

Per Netty 4.0 e successivi, si dovrebbe estendere ChannelDuplexHandler come nell'esempio da 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()); 
      } 
     } 
    } 
}