netty issue when writeAndFlush called from different InboundChannelHandlerAdapter.channelRead

Question

I've got an issue, for which I am unable to post full code (sorry), due to security reasons. The gist of my issue is that I have a ServerBootstrap, created as follows:

bossGroup = new NioEventLoopGroup();
workerGroup = new NioEventLoopGroup();
final ServerBootstrap b = new ServerBootstrap();

b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)                        
.childHandler(new ChannelInitializer<SocketChannel>() {

@Override
public void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, 3000));
                                //Adds the MQTT encoder and decoder
                                ch.pipeline().addLast("decoder", new MyMessageDecoder());
                                ch.pipeline().addLast("encoder", new MyMessageEncoder());
                                ch.pipeline().addLast(createMyHandler());
                            }
                        }).option(ChannelOption.SO_BACKLOG, 128).option(ChannelOption.SO_REUSEADDR, true)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .childOption(ChannelOption.SO_KEEPALIVE, true);

                // Bind and start to accept incoming connections.
                channelFuture = b.bind(listenAddress, listenPort);

With createMyHandlerMethod() that basically returns an extended implementation of ChannelInboundHandlerAdapter

I also have a "client" listener, that listens for incoming connection requests, and is loaded as follows:

    final String host = getHost();
        final int port = getPort();
        nioEventLoopGroup = new NioEventLoopGroup();

        bootStrap = new Bootstrap();
        bootStrap.group(nioEventLoopGroup);
        bootStrap.channel(NioSocketChannel.class);
        bootStrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootStrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, getKeepAliveInterval()));
                ch.pipeline().addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimeoutHandler());
                ch.pipeline().addLast("decoder", new MyMessageDecoder());
                ch.pipeline().addLast("encoder", new MyMessageEncoder());
                ch.pipeline().addLast(MyClientHandler.this);
            }
        })
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.TCP_NODELAY, true);

        // Start the client.
        try {
            channelFuture = bootStrap.connect(host, port).sync();
        } catch (InterruptedException e) {
            throw new MyException(“Exception”, e);
        }

Where MyClientHandler is again a subclassed instance of ChannelInboundHandlerAdapter. Everything works fine, I get messages coming in from the "server" adapter, i process them, and send them back on the same context. And vice-versa for the "client" handler.

The problem happens when I have to (for some messages) proxy them from the server or client handler to other connection. Again, I am very sorry for not being able to post much code, but the gist of it is that I'm calling from:

serverHandler.channelRead(ChannelHandlerContext ctx, Object msg) {
     if (msg instanceof myProxyingMessage) {
        if (ctx.channel().isActive()) {
             ctx.channel().writeAndFlush(someOtherMessage);
             **getClientHandler().writeAndFlush(myProxyingMessage);**
        }
     }
}

Now here's the problem: the bolded (client) writeAndFlush - never actually writes the message bytes, it doesn't throw any errors. The ChannelFuture returns all false (success, cancelled, done). And if I sync on it, eventually it times out for other reasons (connection timeout set within my code).

I know I haven't posted all of my code, but I'm hoping that someone has some tips and/or pointers for how to isolate the problem of WHY it is not writing to the client context. I'm not a Netty expert by any stretch, and most of this code was written by someone else. They are both subclassing ChannelInboundHandlerAdapter

Feel free to ask any questions if you have any.

*****EDIT********* I tried to proxy the request back to a DIFFERENT context/channel (ie, the client channel) using the following test code:

public void proxyPubRec(int messageId) throws MQTTException {
        logger.log(logLevel, "proxying PUBREC to context:  " + debugContext());
        PubRecMessage pubRecMessage = new PubRecMessage();
        pubRecMessage.setMessageID(messageId);
        pubRecMessage.setRemainingLength(2);
        logger.log(logLevel, "pipeline writable flag:  " + ctx.pipeline().channel().isWritable());
        MyMQTTEncoder encoder = new MyMQTTEncoder();

        ByteBuf buff = null;
        try {
            buff = encoder.encode(pubRecMessage);
            ctx.channel().writeAndFlush(buff);

        } catch (Throwable t) {
            logger.log(Level.SEVERE, "unable to encode PUBREC");
        } finally {
            if (buff != null) {
                buff.release();
            }
        }
    }

    public class MyMQTTEncoder extends MQTTEncoder {
        public ByteBuf encode(AbstractMessage msg) {
            PooledByteBufAllocator allocator = new PooledByteBufAllocator();
            ByteBuf buf = allocator.buffer();
            try {
                super.encode(ctx, msg, buf);
            } catch (Throwable t) {
                logger.log(Level.SEVERE, "unable to encode PUBREC, " + t.getMessage());
            }
            return buf;
        }
    }

But the above at line: ctx.channel().writeAndFlush(buff) is NOT writing to the other channel - any tips/tricks on debugging this sort of issue?

Neeraj Gupta Neeraj Gupta · Accepted Answer · 2015-09-16T08:52:26

someOtherMessage has to be ByteBuf.

So, take this :

serverHandler.channelRead(ChannelHandlerContext ctx, Object msg) {
     if (msg instanceof myProxyingMessage) {
        if (ctx.channel().isActive()) {
             ctx.channel().writeAndFlush(someOtherMessage);
             **getClientHandler().writeAndFlush(myProxyingMessage);**
        }
     }
}

... and replace it with this :

serverHandler.channelRead(ChannelHandlerContext ctx, Object msg) {
     if (msg instanceof myProxyingMessage) {
        if (ctx.channel().isActive()) {
             ctx.channel().writeAndFlush(ByteBuf);
             **getClientHandler().writeAndFlush(myProxyingMessage);**
        }
     }
}

netty issue when writeAndFlush called from different InboundChannelHandlerAdapter.channelRead

2 Answers