ServerMain.java

package netty4.example;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.AbstractByteBuf;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.*;
import io.netty.handler.codec.bytes.ByteArrayDecoder;
import io.netty.handler.codec.bytes.ByteArrayEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.internal.logging.InternalLoggerFactory;
import io.netty.util.internal.logging.Slf4JLoggerFactory;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
public class ServerMain extends SimpleChannelInboundHandler<ServerMain.ProtocolMessagePojo>
{
    public static class ProtocolMessagePojo
    {
        public String _message;
    }

    public static class ProtocolEncoderDecoder extends MessageToMessageCodec<byte[], ProtocolMessagePojo>
    {

        @Override
        protected void encode(ChannelHandlerContext _ctx, ProtocolMessagePojo _pojo, List<Object> _list) throws Exception {
            _list.add(_pojo._message.getBytes(StandardCharsets.UTF_8));
        }

        @Override
        protected void decode(ChannelHandlerContext _ctx, byte[] _bytes, List<Object> _list) throws Exception {
            ProtocolMessagePojo _pojo = new ProtocolMessagePojo();
            _pojo._message = new String(_bytes, StandardCharsets.UTF_8);
            _list.add(_pojo);
        }
    }

    public static class DelimiterFrameAppender extends MessageToMessageEncoder<ByteBuf>
    {
        char _a;

        public DelimiterFrameAppender(char _a)
        {
            this._a = _a;
        }

        @Override
        protected void encode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception
        {
            int _len = msg.readableBytes()+1;
            ByteBuf _buf = ctx.alloc().buffer(_len);
            _buf.writeBytes(msg);
            _buf.writeByte((byte) this._a);
            out.add(_buf);
        }
    }

    static ServerMain _instance;
    static Thread _tcpThread;
    static int _tcpPort = 56565;

    public static void main(String[] args) {
        InternalLoggerFactory.setDefaultFactory(new Slf4JLoggerFactory());

        _instance = new ServerMain();
        _tcpThread = new Thread(_instance::runTcp);

        _tcpThread.start();
    }

    NioEventLoopGroup _tcpEventLoopGroup;
    ServerBootstrap _tcpBootstrap;

    @SneakyThrows
    public void runTcp(){
        _tcpEventLoopGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors()*2+1);
        try
        {
            _tcpBootstrap = new ServerBootstrap();
            _tcpBootstrap.group(_tcpEventLoopGroup);
            _tcpBootstrap.channel(NioServerSocketChannel.class);
            _tcpBootstrap.localAddress(new InetSocketAddress(_tcpPort));
            _tcpBootstrap.option(ChannelOption.SO_BACKLOG, 128);
            _tcpBootstrap.option(ChannelOption.SO_REUSEADDR, true);
            _tcpBootstrap.option(ChannelOption.TCP_NODELAY, true);
            _tcpBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {
                protected void initChannel(NioSocketChannel ch) throws Exception
                {
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.WARN));
                    // server input
                    //ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2));
                    ch.pipeline().addLast(new DelimiterBasedFrameDecoder(64, true, ch.alloc().buffer(1).writeByte('\n')));
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.WARN));
                    ch.pipeline().addLast(new ByteArrayDecoder());
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.WARN));
                    // server output
                    //ch.pipeline().addLast(new LengthFieldPrepender(2));
                    ch.pipeline().addLast(new DelimiterFrameAppender('\n'));
                    ch.pipeline().addLast(new ByteArrayEncoder());
                    ch.pipeline().addLast(new ProtocolEncoderDecoder());

                    // pojo codec
                    ch.pipeline().addLast(ServerMain.this);
                }
            });
            ChannelFuture channelFuture = _tcpBootstrap.bind().sync();
        }
        catch(Exception _e)
        {
            log.error("ERROR", _e);
        }
        finally
        {
            //_tcpEventLoopGroup.shutdownGracefully().sync();
        }
    }


    @Override
    protected void channelRead0(ChannelHandlerContext _ctx, ProtocolMessagePojo _pojo) throws Exception
    {
        log.warn("POJO: "+_pojo._message);
        _ctx.channel().write(_pojo);
        _ctx.channel().flush();
    }

}