@Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast("messageDecoder",
                new LengthFieldBasedFrameDecoder(this.maxMsgLength,
                        0, 4, 0, 0, true));
        ch.pipeline().addLast("readTimeoutHandler",
                new ReadTimeoutHandler(DEFAULT_READ_IDLE_TIME, TimeUnit.MILLISECONDS));

        if (processor != null) {
            try {
                Class<? extends ChannelInboundHandlerAdapter> clazz =
                        (Class<? extends ChannelInboundHandlerAdapter>) Class.forName(messageHandlerName);

                Constructor<?> ctor =
                        clazz.getConstructor(AbstractSource.class, ServiceDecoder.class,
                                ChannelGroup.class, Integer.class, Long.class);

                ChannelInboundHandlerAdapter messageHandler = (ChannelInboundHandlerAdapter) ctor
                        .newInstance(source, serviceDecoder, allChannels, maxConnections, msgValidThresholdDays);

                ch.pipeline().addLast("messageHandler", messageHandler);
            } catch (Exception e) {
                LOG.info("SimpleChannelHandler.newInstance  has error:" + name, e);
            }
        }
    }
get server factory

@param source
@param allChannels
@param serviceDecoder
@param messageHandlerName
@param maxMsgLength
@param maxCons
@param name
inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/source/ServerMessageFactory.java