netty - 使用 Netty 4，我如何编写一个可以定期注入消息并仍然传递消息的处理程序？

Question

我正在尝试编写一个每分钟将消息注入通道的类。我已经想出了如何使用下面的代码来完成此操作，但我认为我的刷新方法是错误的。刷新上游消息后，我注意到套接字立即关闭。

public class Pinger extends ChannelOutboundMessageHandlerAdapter<ByteBuf> {
    private static final ByteBuf DUMMY = Unpooled.wrappedBuffer("DUMMY".getBytes());

    @Override
    public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception{
        super.connect(ctx, remoteAddress, localAddress, promise);

        ctx.executor().scheduleAtFixedRate(new RepeatTask(ctx), 0, 60, TimeUnit.SECONDS);
    }

    private final class RepeatTask implements Runnable {
        private final ChannelHandlerContext ctx;

        public RepeatTask(ChannelHandlerContext ctx){
            this.ctx = ctx;
        }

        public void run() {
            if(ctx.channel().isActive()){
                ctx.write(DUMMY.copy());
            }
        }
    }

    @Override
    public void flush(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        ctx.nextOutboundMessageBuffer().add(msg);
        ctx.flush();
    }
}

我还要注意，这个处理程序位于复杂管道的中间。

Answer 1

我想我想出了如何使这项工作。ChannelStateHandlerAdapter 是一个更好的继承类。

public class Pinger extends ChannelStateHandlerAdapter {
    private static final ByteBuf DUMMY = Unpooled.wrappedBuffer("DUMMY".getBytes());

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();
        ctx.executor().scheduleAtFixedRate(new PingTask(ctx), 0, 60, TimeUnit.SECONDS);
    }

    private final class RepeatTask implements Runnable {
        private final ChannelHandlerContext ctx;

        public RepeatTask(ChannelHandlerContext ctx){
            this.ctx = ctx;
        }

        public void run() {
            if(ctx.channel().isActive()){
                ctx.write(DUMMY.copy());
            }
        }
    }

    @Override
    public void inboundBufferUpdated(ChannelHandlerContext ctx)
            throws Exception {
        ctx.fireInboundBufferUpdated();
    }
}