0

I am trying to implement a TCP server in java using Netty. I am able to handle message of length < 1024 correctly but when I receive message more than 1024, I am only able to see partial message.

I did some research, found that I should implement replayingdecoder but I am unable to understand how to implement the decode method

My message uses JSON

Netty version 4.1.27

protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception 

My Server setup

    EventLoopGroup group;

    group = new NioEventLoopGroup(this.numThreads);

    try {
        ServerBootstrap serverBootstrap;
        RequestHandler requestHandler;
        ChannelFuture channelFuture;

        serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(group);
        serverBootstrap.channel(NioServerSocketChannel.class);
        serverBootstrap.localAddress(new InetSocketAddress("::", this.port));

        requestHandler = new RequestHandler(this.responseManager, this.logger);

        serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                socketChannel.pipeline().addLast(requestHandler);
            }
        });

        channelFuture = serverBootstrap.bind().sync();
        channelFuture.channel().closeFuture().sync();
    }
    catch(Exception e){
        this.logger.info(String.format("Unknown failure %s", e.getMessage()));
    }
    finally {
        try {
            group.shutdownGracefully().sync();
        }
        catch (InterruptedException e) {
            this.logger.info(String.format("Error shutting down %s", e.getMessage()));
        }

    }

My current request handler

package me.chirag7jain.Response;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;
import org.apache.logging.log4j.Logger;

import java.net.InetSocketAddress;

public class RequestHandler extends ChannelInboundHandlerAdapter {

private ResponseManager responseManager;
private Logger logger;

public RequestHandler(ResponseManager responseManager, Logger logger) {
    this.responseManager = responseManager;
    this.logger = logger;
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ByteBuf byteBuf;
    String data, hostAddress;

    byteBuf = (ByteBuf) msg;
    data = byteBuf.toString(CharsetUtil.UTF_8);
    hostAddress = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress().getHostAddress();

    if (!data.isEmpty()) {
        String reply;

        this.logger.info(String.format("Data received %s from %s", data, hostAddress));
        reply = this.responseManager.reply(data);

        if (reply != null) {
            ctx.write(Unpooled.copiedBuffer(reply, CharsetUtil.UTF_8));
        }
    }
    else {
        logger.info(String.format("NO Data received from %s", hostAddress));
    }
}

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    this.logger.info(String.format("Received Exception %s", cause.getMessage()));
    ctx.close();
}

2 Answers 2

2
+100

I would accept data in channelRead() and accumulate it in a buffer. Before return from channelRead() I would invoke read() on a Channel. You may need to record other data, as per your needs. When netty invokes channelReadComplete(), there is a moment to send whole buffer to your ResponseManager.

Channel read(): Request to Read data from the Channel into the first inbound buffer, triggers an ChannelInboundHandler.channelRead(ChannelHandlerContext, Object) event if data was read, and triggers a channelReadComplete event so the handler can decide to continue reading.

Your Channel object is accessible by ctx.channel().

Try this code:

private final AttributeKey<StringBuffer> dataKey = AttributeKey.valueOf("dataBuf");

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    ByteBuf byteBuf;
    String data, hostAddress;

    StringBuffer dataBuf = ctx.attr(dataKey).get();
    boolean allocBuf = dataBuf == null;
    if (allocBuf) dataBuf = new StringBuffer();

    byteBuf = (ByteBuf) msg;
    data = byteBuf.toString(CharsetUtil.UTF_8);
    hostAddress = ((InetSocketAddress) ctx.channel().remoteAddress()).getAddress().getHostAddress();

    if (!data.isEmpty()) {
        this.logger.info(String.format("Data received %s from %s", data, hostAddress));
    }
    else {
        logger.info(String.format("NO Data received from %s", hostAddress));
    }

    dataBuf.append(data);
    if (allocBuf) ctx.attr(dataKey).set(dataBuf);

    ctx.channel().read();
}

@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    StringBuffer dataBuf = ctx.attr(dataKey).get();
    if (dataBuf != null) {
        String reply;

        reply = this.responseManager.reply(dataBuf.toString());

        if (reply != null) {
            ctx.write(Unpooled.copiedBuffer(reply, CharsetUtil.UTF_8));
        }
    }
    ctx.attr(dataKey).set(null);

    ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
}
2
  • Didn't really understand it Commented Jul 29, 2018 at 23:24
  • See edit. You may need to tweak it a bit, but there you have all you need.
    – Jomu
    Commented Jul 30, 2018 at 7:01
0

An application protocol with variable length messages must have:

  • a length word
  • a terminator character or sequence, which in turn implies an escape character in case the data contains the terminator
  • a self-describing protocol such as XML.
2
  • 1
    I am using JSON Commented Jul 27, 2018 at 0:01
  • Does not only have to have these. You can write a VARINT in order to prepend the payload length. Var Int Sample . He could do this in the replaying decoder or his own custom handler. I personally implement a packet id protocol that the packet id tells me the payload length field type; I.E byte, short, medium or varint, then read that field, then the data. Commented Jul 27, 2018 at 0:39

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.