Netty channel write not reaching handlers

Question

I'm learning Netty and prototyping a simple app which sends an object over TCP. My issue is that when I call Channel.write from the server side with my message, it doesn't seem to reach the handlers in the pipeline. When I send a message from client to server, it works as expected.

Here's the code.

The server:

public class Main {     
    private int serverPort;

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;

    private ServerBootstrap boot;
    private ChannelFuture future;

    private SomeDataChannelDuplexHandler duplex;

    private Channel ch;

    public Main(int serverPort) {
        this.serverPort = serverPort;
    }

    public void initialise() {      
        boot = new ServerBootstrap();       
        bossGroup = new NioEventLoopGroup();
        workerGroup = new NioEventLoopGroup();

        boot.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(0, 0, 2));

                    // Inbound
                    ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 0));
                    ch.pipeline().addLast(new SomeDataDecoder());

                    // Outbound
                    ch.pipeline().addLast(new LengthFieldPrepender(2));
                    ch.pipeline().addLast(new SomeDataEncoder()); 

                    // In-Out
                    ch.pipeline().addLast(new SomeDataChannelDuplexHandler());
                }
            })      
            .option(ChannelOption.SO_BACKLOG, 128)
            .childOption(ChannelOption.SO_KEEPALIVE, true); 
    }

    public void sendMessage() { 
        SomeData fd = new SomeData("hello", "localhost", 1234);     
        ChannelFuture future = ch.writeAndFlush(fd);        
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    System.out.println("send error: " + future.cause().toString());
                } else {
                    System.out.println("send message ok");  
                }
            }
        });
    }

    public void startServer(){
        try {
            future = boot.bind(serverPort)
                    .sync()
                    .addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture future) throws Exception {
                            ch = future.channel();
                        }
            });
        } catch (InterruptedException e) {
            // log failure
        }
    }

    public void stopServer() {
        workerGroup.shutdownGracefully()
            .addListener(e -> System.out.println("workerGroup shutdown"));

        bossGroup.shutdownGracefully()
            .addListener(e -> System.out.println("bossGroup shutdown"));
    }

    public static void main(String[] args) throws InterruptedException {

        Main m = new Main(5000);

        m.initialise();
        m.startServer();

        final Scanner scanner = new Scanner(System.in);

        System.out.println("running.");

        while (true) {

            final String input = scanner.nextLine();

            if ("q".equals(input.trim())) {
                break;
            } else {
                m.sendMessage();
            }
        }

        scanner.close();
        m.stopServer();
    }
}

The duplex channel handler:

public class SomeDataChannelDuplexHandler extends ChannelDuplexHandler {

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("duplex channel active");
        ctx.fireChannelActive();
    }

    @Override 
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
        System.out.println("duplex channelRead");
        if (msg instanceof SomeData) {
            SomeData sd = (SomeData) msg;
            System.out.println("received: " + sd);
        } else {
            System.out.println("some other object");
        }
        ctx.fireChannelRead(msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.ALL_IDLE) { // idle for no read and write
                System.out.println("idle: " + event.state());
            }
        }
    }   
}

And finally the encoder (the decoder is similar):

public class SomeDataEncoder extends MessageToByteEncoder<SomeData> {

    @Override
    protected void encode(ChannelHandlerContext ctx, SomeData msg, ByteBuf out) throws Exception {

        System.out.println("in encoder, msg = " + msg);
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(bos);

        oos.writeObject(msg.getName());
        oos.writeObject(msg.getIp());
        oos.writeInt(msg.getPort());
        oos.close();

        byte[] serialized = bos.toByteArray();
        int size = serialized.length;

        ByteBuf encoded = ctx.alloc().buffer(size);
        encoded.writeBytes(bos.toByteArray());

        out.writeBytes(encoded);
    }
}

The client side:

public class Client {

    String host = "10.188.36.66";
    int port = 5000;

    EventLoopGroup workerGroup = new NioEventLoopGroup();
    ChannelFuture f;
    private Channel ch;

    public Client() {
    }

    public void startClient() throws InterruptedException {
        Bootstrap boot = new Bootstrap();
        boot.group(workerGroup);
        boot.channel(NioSocketChannel.class);
        boot.option(ChannelOption.SO_KEEPALIVE, true);
        boot.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {            
                // Inbound
                ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 0));
                ch.pipeline().addLast(new SomeDataDecoder()); 

                // Outbound
                ch.pipeline().addLast(new LengthFieldPrepender(2));
                ch.pipeline().addLast(new SomeDataEncoder());

                // Handler
                ch.pipeline().addLast(new SomeDataHandler());
            }
        });

        // Start the client
        f = boot.connect(host, port).sync();
        f.addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture future) throws Exception {
                System.out.println("connected to server");
                ch = f.channel();
            }
        });
    }

    public void stopClient() {      
        workerGroup.shutdownGracefully();
    }

    private void writeMessage(String input) {
        SomeData data = new SomeData("client", "localhost", 3333);
        ChannelFuture fut = ch.writeAndFlush(data);
        fut.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                System.out.println("send message");
            }
        });
    }

    public static void main(String[] args) throws InterruptedException {
        Client client = new Client();
        client.startClient();        

        System.out.println("running.\n\n");
        final Scanner scanner = new Scanner(System.in);

        while (true) {

            final String input = scanner.nextLine();

            if ("q".equals(input.trim())) {
                break;
            } else {
                client.writeMessage(input);
            }   
        }   

        scanner.close();
        client.stopClient();  //call this at some point to shutdown the client
    }

}

and the handler:

public class SomeDataHandler extends SimpleChannelInboundHandler<SomeData> {

    private ChannelHandlerContext ctx;

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("connected");
        this.ctx = ctx;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, SomeData msg) throws Exception {
        System.out.println("got message: " + msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {   
        System.out.println("caught exception: " + cause.getMessage());
        ctx.close();
    }
}

When I send a message via the console on the server side, I get the output:

running.
duplex channel active
duplex read
idle: ALL_IDLE
idle: ALL_IDLE

send message ok

So it looks as though the message is sent but nothing is received on the client side.

When I do it from the client side I get (on the server console):

in decoder, numBytes in message = 31
duplex channelRead
received: SomeData [name=client, ip=localhost, port=3333]

which is what I expect.

So where's the problem? Is it something to do with using a ChannelDuplexHandler on the server side and a SimpleChannelInboundHandler on the client side? Is there something I need to call to kick the message down the pipeline?

UPDATE I've added a check for future.isSuccess() in the server sendMessage method and I get send error: java.lang.UnsupportedOperationException on the console.

Answer 1

(Posted on behalf of the OP).

For anyone who's interested, the problem was that I was trying to send the message on the server channel and not the normal channel. This post pointed me in the right direction.