@Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            logger.debug("server message receive!");
            if (!(msg instanceof RpcDataPack)) {
                return;
            }
            logger.debug("server RpcDataPack message receive!");
            RpcDataPack dataPack = (RpcDataPack) msg;
            RPCProtos.RpcConnHeader connHeader;
            RPCProtos.RequestHeader requestHeader;
            RPCProtos.RequestBody rpcRequestBody;
            int rmtVersion = RpcProtocol.RPC_PROTOCOL_VERSION;
            Channel channel = ctx.channel();
            if (channel == null) {
                return;
            }
            String rmtaddrIp = getRemoteAddressIP(channel);
            try {
                if (!isServiceStarted()) {
                    throw new ServerNotReadyException("RpcServer is not running yet");
                }
                List<ByteBuffer> req = dataPack.getDataLst();
                ByteBufferInputStream dis = new ByteBufferInputStream(req);
                connHeader = RPCProtos.RpcConnHeader.parseDelimitedFrom(dis);
                requestHeader = RPCProtos.RequestHeader.parseDelimitedFrom(dis);
                rmtVersion = requestHeader.getProtocolVer();
                rpcRequestBody = RPCProtos.RequestBody.parseDelimitedFrom(dis);
            } catch (Throwable e1) {
                if (!(e1 instanceof ServerNotReadyException)) {
                    if (rmtaddrIp != null) {
                        AtomicLong count = errParseAddrMap.get(rmtaddrIp);
                        if (count == null) {
                            AtomicLong tmpCount = new AtomicLong(0);
                            count = errParseAddrMap.putIfAbsent(rmtaddrIp, tmpCount);
                            if (count == null) {
                                count = tmpCount;
                            }
                        }
                        count.incrementAndGet();
                        long befTime = lastParseTime.get();
                        long curTime = System.currentTimeMillis();
                        if (curTime - befTime > 180000) {
                            if (lastParseTime.compareAndSet(befTime, System.currentTimeMillis())) {
                                logger.warn(new StringBuilder(512)
                                        .append("[Abnormal Visit] Abnormal Message Content visit list is :")
                                        .append(errParseAddrMap).toString());
                                errParseAddrMap.clear();
                            }
                        }
                    }
                }
                List<ByteBuffer> res =
                        prepareResponse(null, rmtVersion, RPCProtos.ResponseHeader.Status.FATAL,
                                e1.getClass().getName(), new StringBuilder(512)
                                        .append("IPC server unable to read call parameters:")
                                        .append(e1.getMessage()).toString());
                if (res != null) {
                    dataPack.setDataLst(res);
                    channel.writeAndFlush(dataPack);
                }
                return;
            }
            try {
                RequestWrapper requestWrapper =
                        new RequestWrapper(requestHeader.getServiceType(),
                                this.protocolType, requestHeader.getProtocolVer(),
                                connHeader.getFlag(), rpcRequestBody.getTimeout());
                requestWrapper.setMethodId(rpcRequestBody.getMethod());
                requestWrapper.setRequestData(PbEnDecoder.pbDecode(true,
                        rpcRequestBody.getMethod(), rpcRequestBody.getRequest().toByteArray()));
                requestWrapper.setSerialNo(dataPack.getSerialNo());
                RequestContext context =
                        new NettyRequestContext(requestWrapper, ctx, System.currentTimeMillis());
                protocols.get(this.protocolType).handleRequest(context, rmtaddrIp);
            } catch (Throwable ee) {
                List<ByteBuffer> res =
                        prepareResponse(null, rmtVersion, RPCProtos.ResponseHeader.Status.FATAL,
                                ee.getClass().getName(), new StringBuilder(512)
                                        .append("IPC server handle request error :")
                                        .append(ee.getMessage()).toString());
                if (res != null) {
                    dataPack.setDataLst(res);
                    ctx.channel().writeAndFlush(dataPack);
                }
                return;
            }
        }
Invoked when a message object was received
from a remote peer.
inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyRpcServer.java