Java 类io.netty.handler.codec.LengthFieldBasedFrameDecoder 实例源码

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    if (enableTLS) {
        File tlsCert = new File(serviceConfig.getTlsCertificateFilePath());
        File tlsKey = new File(serviceConfig.getTlsKeyFilePath());
        SslContextBuilder builder = SslContextBuilder.forServer(tlsCert, tlsKey);
        if (serviceConfig.isTlsAllowInsecureConnection()) {
            builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
        } else {
            if (serviceConfig.getTlsTrustCertsFilePath().isEmpty()) {
                // Use system default
                builder.trustManager((File) null);
            } else {
                File trustCertCollection = new File(serviceConfig.getTlsTrustCertsFilePath());
                builder.trustManager(trustCertCollection);
            }
        }
        SslContext sslCtx = builder.clientAuth(ClientAuth.OPTIONAL).build();
        ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
    }
    ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(PulsarDecoder.MaxFrameSize, 0, 4, 0, 4));
    ch.pipeline().addLast("handler", new ServerConnection(discoveryService));
}

public void run() {
    workerGroup = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(workerGroup);
        b.channel(NioSocketChannel.class);
        // b.option(ChannelOption.SO_KEEPALIVE, true);
        b.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4));
                pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                pipeline.addLast("decoder", new MsgPackDecode());
                pipeline.addLast("encoder", new MsgPackEncode());
                pipeline.addLast(new ClientHandler());
            }
        });
        channel = b.connect(clientProperties.getServerHost(), clientProperties.getServerPort()).sync().channel();
        status = Status.START;
        channel.closeFuture().sync();
    } catch (Exception e) {
        e.printStackTrace();
    }
    status = Status.STOP;
}

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pip = ch.pipeline();
    int maxLength = 1048576;
    int lengthFieldLength = 4;
    int ignoreLength = -4;
    int offset = 0;
    pip.addLast(new LengthFieldBasedFrameDecoder(maxLength, offset, lengthFieldLength, ignoreLength, lengthFieldLength));
    pip.addLast(new MessageDecoder(builder.getImessageandhandler()));
    pip.addLast(new LengthFieldPrepender(4, true));
    pip.addLast(new MessageEncoder(builder.getImessageandhandler()));
    pip.addLast(new MessageExecutor(builder.getConsumer(), builder.getListener()));
    for (ChannelHandler handler : builder.getExtraHandlers()) {
        pip.addLast(handler);
    }
}

/**
 * Init channel.
 */
public void initChannel(Channel ch) {
    ChannelPipeline cp = ch.pipeline();
    //
    cp.addLast("frame", new LengthFieldBasedFrameDecoder(this.maxFrameLength, 0, lengthFieldLength, 0, lengthFieldLength));
    cp.addLast("prepender", this.prepender);
    if (isAutoConnect()) {
        cp.addLast("autoConnect", new AutoConnectHandler(this));
    }
    //  Event Handler
    if (!handlers.isEmpty()) {
        for (Pair<String, ChannelHandler> pair : handlers) {
            cp.addLast(pair.getLeft(), pair.getRight());
        }
    }
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    if (enableTLS) {
        File tlsCert = new File(serviceConfig.getTlsCertificateFilePath());
        File tlsKey = new File(serviceConfig.getTlsKeyFilePath());
        SslContextBuilder builder = SslContextBuilder.forServer(tlsCert, tlsKey);
        if (serviceConfig.isTlsAllowInsecureConnection()) {
            builder.trustManager(InsecureTrustManagerFactory.INSTANCE);
        } else {
            if (serviceConfig.getTlsTrustCertsFilePath().isEmpty()) {
                // Use system default
                builder.trustManager((File) null);
            } else {
                File trustCertCollection = new File(serviceConfig.getTlsTrustCertsFilePath());
                builder.trustManager(trustCertCollection);
            }
        }
        SslContext sslCtx = builder.clientAuth(ClientAuth.OPTIONAL).build();
        ch.pipeline().addLast(TLS_HANDLER, sslCtx.newHandler(ch.alloc()));
    }
    ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(PulsarDecoder.MaxFrameSize, 0, 4, 0, 4));
    ch.pipeline().addLast("handler", new ServerCnx(brokerService));
}

public void startMockBrokerService() throws Exception {
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("mock-pulsar-%s").build();
    final int numThreads = 2;

    final int MaxMessageSize = 5 * 1024 * 1024;

    try {
        workerGroup = EventLoopUtil.newEventLoopGroup(numThreads, threadFactory);

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(workerGroup, workerGroup);
        bootstrap.channel(EventLoopUtil.getServerSocketChannelClass(workerGroup));
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(MaxMessageSize, 0, 4, 0, 4));
                ch.pipeline().addLast("handler", new MockServerCnx());
            }
        });
        // Bind and start to accept incoming connections.
        bootstrap.bind(brokerServicePort).sync();
    } catch (Exception e) {
        throw e;
    }
}

public static void service() throws Exception {
    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workerGroup);
    bootstrap.channel(NioServerSocketChannel.class);
    bootstrap.childHandler(new ChannelInitializer<Channel>() {

        @Override
        protected void initChannel(Channel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast(new LengthFieldPrepender(4));
            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast(new TcpServerHandler());
        }

    });
    ChannelFuture f = bootstrap.bind(IP, PORT).sync();
    f.channel().closeFuture().sync();
    System.out.println("TCP服务器已启动");
}

/**
 * 初始化Bootstrap
 * @return
 */
public static final Bootstrap getBootstrap(){
    EventLoopGroup group = new NioEventLoopGroup();
    Bootstrap b = new Bootstrap();
    b.group(group).channel(NioSocketChannel.class);
    b.handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
            pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast("handler", new TcpClientHandler());
        }
    });
    b.option(ChannelOption.SO_KEEPALIVE, true);
    return b;
}

protected static void run() throws Exception {
    ServerBootstrap b = new ServerBootstrap();
    b.group(bossGroup, workerGroup);
    b.channel(NioServerSocketChannel.class);
    b.childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
            pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast(new TcpServerHandler());
        }
    });

    b.bind(IP, PORT).sync();
    System.out.println("TCP服务器已启动");
}

@Provides
@EppProtocol
static ImmutableList<Provider<? extends ChannelHandler>> provideHandlerProviders(
    Provider<SslServerInitializer<NioSocketChannel>> sslServerInitializerProvider,
    Provider<ProxyProtocolHandler> proxyProtocolHandlerProvider,
    @EppProtocol Provider<ReadTimeoutHandler> readTimeoutHandlerProvider,
    Provider<LengthFieldBasedFrameDecoder> lengthFieldBasedFrameDecoderProvider,
    Provider<LengthFieldPrepender> lengthFieldPrependerProvider,
    Provider<EppServiceHandler> eppServiceHandlerProvider,
    Provider<LoggingHandler> loggingHandlerProvider,
    Provider<FullHttpRequestRelayHandler> relayHandlerProvider) {
  return ImmutableList.of(
      proxyProtocolHandlerProvider,
      sslServerInitializerProvider,
      readTimeoutHandlerProvider,
      lengthFieldBasedFrameDecoderProvider,
      lengthFieldPrependerProvider,
      eppServiceHandlerProvider,
      loggingHandlerProvider,
      relayHandlerProvider);
}

@Test
public void testFailSlowTooLongFrameRecovery() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(
            new LengthFieldBasedFrameDecoder(5, 0, 4, 0, 4, false));

    for (int i = 0; i < 2; i ++) {
        assertFalse(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 2 })));
        try {
            assertTrue(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0 })));
            fail(DecoderException.class.getSimpleName() + " must be raised.");
        } catch (TooLongFrameException e) {
            // Expected
        }

        ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 1, 'A' }));
        ByteBuf buf = releaseLater((ByteBuf) ch.readInbound());
        assertEquals("A", buf.toString(CharsetUtil.ISO_8859_1));
        buf.release();
    }
}

@Test
public void testFailFastTooLongFrameRecovery() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(
            new LengthFieldBasedFrameDecoder(5, 0, 4, 0, 4));

    for (int i = 0; i < 2; i ++) {
        try {
            assertTrue(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 2 })));
            fail(DecoderException.class.getSimpleName() + " must be raised.");
        } catch (TooLongFrameException e) {
            // Expected
        }

        ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 0, 0, 1, 'A' }));
        ByteBuf buf = releaseLater((ByteBuf) ch.readInbound());
        assertEquals("A", buf.toString(CharsetUtil.ISO_8859_1));
        buf.release();
    }
}

public void connect() throws Exception {
    workerGroup = new NioEventLoopGroup();

    Bootstrap bootstrap =
            new Bootstrap()
                    .group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();

                            pipeline.addLast(new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
                            pipeline.addLast(new ProtobufDecoder(Protocol.BaseMessage.getDefaultInstance()));

                            clientHandler = new ClientHandler();
                            pipeline.addLast(clientHandler);

                            pipeline.addLast(new LengthFieldPrepender(4));
                            pipeline.addLast(new ProtobufEncoder());
                        }
                    });

    ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
    channel = channelFuture.channel();
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
  LOGGER.info("Setting up Server channel !!");
  ch.pipeline().addLast("decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
  ch.pipeline().addLast("encoder", new LengthFieldPrepender(4));
  //ch.pipeline().addLast("logger", new LoggingHandler());
  // Create server metric for this handler and add to aggregate if present
  NettyServerMetrics serverMetric =
      new NettyServerMetrics(_registry, NettyTCPServer.class.getName() + "_" + Utils.getUniqueId() + "_");

  if (null != _globalMetrics) {
    _globalMetrics.addTransportClientMetrics(serverMetric);
  }

  ch.pipeline().addLast("request_handler",
      new NettyChannelInboundHandler(_handlerFactory.createNewRequestHandler(), serverMetric, _defaultLargeQueryLatencyMs));
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {

    if (sslContext != null) {
        ch.pipeline().addLast(sslContext.newHandler(ch.alloc(), host, port));
    }

    // In
    ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4));
    if (cryptoFunction != null) {
        ch.pipeline().addLast(cryptoFunction.getDecoder());
    }
    ch.pipeline().addLast(new PacketDecoder(protocol));

    // Out
    ch.pipeline().addLast(new LengthFieldPrepender(4));
    if (cryptoFunction != null) {
        ch.pipeline().addLast(cryptoFunction.getEncoder());
    }
    ch.pipeline().addLast(new PacketEncoder(protocol));

    // Handler
    ch.pipeline().addLast(new CascadeSession(ch, protocol, sessionListener));
}

@Override
    protected ChannelHandler newChannelInitializer() {
        return new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) throws Exception {
                ChannelPipeline cp = ch.pipeline();
                cp.addLast("frame", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 2, 0, 2));
                cp.addLast("prepender", FRAME_PREPENDER);
                cp.addLast("decoder", GPB_DECODER_HANDLER);
                cp.addLast("encoder", GPB_ENCODER_HANDLER);
                // handler
                cp.addLast("handler", serverHandler);
//                cp.addLast("handler", new ServerHandler());
            }
        };
    }

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();
    // Enable TCPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    p.addLast("logger", new LoggingHandler(LogLevel.DEBUG));

    p.addLast(new LengthFieldBasedFrameDecoder(ByteOrder.LITTLE_ENDIAN, Integer.MAX_VALUE, 0, 4, 0, 4, true));
    p.addLast(new ByteBufferDecoder());

    p.addLast(new LengthFieldPrepender(4, 0, false) {
        @Override
        protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
            ByteBuf outWithLittleEndian = out.order(ByteOrder.LITTLE_ENDIAN);
            super.encode(ctx, msg, outWithLittleEndian);
        }
    });
    p.addLast(new ByteBufferEncoder());
    p.addLast(handler);
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();
    // Enable TCPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    p.addLast("logger", new LoggingHandler(LogLevel.DEBUG));

    p.addLast(new LengthFieldBasedFrameDecoder(ByteOrder.LITTLE_ENDIAN, Integer.MAX_VALUE, 0, 4, 0, 4, true));
    p.addLast(new ByteBufferDecoder());

    p.addLast(new LengthFieldPrepender(4, 0, false) {
        @Override
        protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
            ByteBuf outWithLittleEndian = out.order(ByteOrder.LITTLE_ENDIAN);
            super.encode(ctx, msg, outWithLittleEndian);
        }
    });
    p.addLast(new ByteBufferEncoder());
    p.addLast(new TcpHandler(transportFactory, path, connectionListener));
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();

    // Add SSL handler first to encrypt and decrypt everything.
    // In this example, we use a bogus certificate in the server side
    // and accept any invalid certificates in the client side.
    // You will need something more complicated to identify both
    // and server in the real world.

    SSLEngine engine =
        SecureSocketSslContextFactory.getClientContext().createSSLEngine();
    engine.setUseClientMode(true);

    pipeline.addLast("ssl", new SslHandler(engine));
    // On top of the SSL handler, add the text line codec.
    // pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
    pipeline.addLast("length-decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
    pipeline.addLast("String-decoder", new StringDecoder());  
    pipeline.addLast("length-encoder", new LengthFieldPrepender(4));
    pipeline.addLast("String-encoder", new StringEncoder()); 
    pipeline.addLast("handler", new SecureSocketClientHandler());
}

public void run(String host, int port) throws Exception {

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .remoteAddress(host, port)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(100, 0, 4, 0, 4))
                                    .addLast(new StringDecoder())
                                    .addLast(new LengthFieldClientHandler());
                        }
                    });

            ChannelFuture cf = bootstrap.connect().sync();
            cf.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

private void run(int port) throws InterruptedException, IOException {
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();

    try {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO))
                                .addLast(new LengthFieldBasedFrameDecoder(200, 0, 4, 0, 4))
                                .addLast(new StringDecoder())
                                .addLast(new JsonStringToObjectDecoder())
                                .addLast(new ObjectEchoServerHandler());
                    }
                });
        bootstrap.bind(port).sync().channel().closeFuture().sync();
    } finally {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

}

@Test
public void testFailSlowTooLongFrameRecovery() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(
            new LengthFieldBasedFrameDecoder(5, 0, 4, 0, 4, false));

    for (int i = 0; i < 2; i ++) {
        assertFalse(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 2 })));
        try {
            assertTrue(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0 })));
            fail(DecoderException.class.getSimpleName() + " must be raised.");
        } catch (TooLongFrameException e) {
            // Expected
        }

        ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 1, 'A' }));
        ByteBuf buf = releaseLater((ByteBuf) ch.readInbound());
        assertEquals("A", buf.toString(CharsetUtil.ISO_8859_1));
        buf.release();
    }
}

@Test
public void testFailFastTooLongFrameRecovery() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(
            new LengthFieldBasedFrameDecoder(5, 0, 4, 0, 4));

    for (int i = 0; i < 2; i ++) {
        try {
            assertTrue(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 2 })));
            fail(DecoderException.class.getSimpleName() + " must be raised.");
        } catch (TooLongFrameException e) {
            // Expected
        }

        ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 0, 0, 1, 'A' }));
        ByteBuf buf = releaseLater((ByteBuf) ch.readInbound());
        assertEquals("A", buf.toString(CharsetUtil.ISO_8859_1));
        buf.release();
    }
}

public TCPServerHandler(EventLoopGroup bossGroup, EventLoopGroup workerGroup, ThrowableNioEventLoopGroup eventExecutor, List<Service> services, InetSocketAddress serverAddress) throws InterruptedException {
    ChannelFuture bind = new ServerBootstrap()
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.AUTO_READ, false)
            .option(ChannelOption.SO_BACKLOG, 100)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
                    p.addLast("packetDecoder", new PacketDecoder());
                    p.addLast("frameEncoder", new LengthFieldPrepender(Integer.BYTES));
                    p.addLast("packetEncoder", new PacketEncoder());
                    p.addLast(new ServerChannelAdapter(services, eventExecutor));
                }
            }).bind(serverAddress);

    server = bind.sync()
            .addListener(future -> {
                if (!future.isSuccess()) {
                    LOGGER.error("Failed to bind {}", bind.channel().localAddress());
                }
            }).awaitUninterruptibly().channel();
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ChannelPipeline pipeline = socketChannel.pipeline();

    /* Use Netty's prebuilt tools to handle frame separation on incoming data. */
    pipeline.addLast("lengthDecoder", new LengthFieldBasedFrameDecoder(OPENFLOW_MAXIMUM_FRAME, 2, 2, -4, 0));

    /* Process OpenFlow packets. */
    pipeline.addLast("openflowDecoder", new OpenFlowDecoder());
    pipeline.addLast("openflowEncoder", new OpenFlowEncoder());

    /* Idle Handler, prevent a hung switch or controller from disrupting traffic.  */
    pipeline.addLast("idleStateHandler", new IdleStateHandler(proxy.getIdleReadTimeout(), proxy.getIdleWriteTimeout(), 0, TimeUnit.MILLISECONDS));

    /* OpenFlow Processor. */
    if (downstream) {
        pipeline.addLast("messageHandler", new OpenFlowChannelInboundDownstreamHandler(proxy));
    } else {
        pipeline.addLast("messageHandler", new OpenFlowChannelInboundUpstreamHandler(proxy));
    }
}

@Test
public void testFailSlowTooLongFrameRecovery() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(
            new LengthFieldBasedFrameDecoder(5, 0, 4, 0, 4, false));

    for (int i = 0; i < 2; i ++) {
        assertFalse(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 2 })));
        try {
            assertTrue(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0 })));
            fail(DecoderException.class.getSimpleName() + " must be raised.");
        } catch (TooLongFrameException e) {
            // Expected
        }

        ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 1, 'A' }));
        ByteBuf buf = releaseLater((ByteBuf) ch.readInbound());
        assertEquals("A", buf.toString(CharsetUtil.ISO_8859_1));
        buf.release();
    }
}

@Test
public void testFailFastTooLongFrameRecovery() throws Exception {
    EmbeddedChannel ch = new EmbeddedChannel(
            new LengthFieldBasedFrameDecoder(5, 0, 4, 0, 4));

    for (int i = 0; i < 2; i ++) {
        try {
            assertTrue(ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 2 })));
            fail(DecoderException.class.getSimpleName() + " must be raised.");
        } catch (TooLongFrameException e) {
            // Expected
        }

        ch.writeInbound(Unpooled.wrappedBuffer(new byte[] { 0, 0, 0, 0, 0, 1, 'A' }));
        ByteBuf buf = releaseLater((ByteBuf) ch.readInbound());
        assertEquals("A", buf.toString(CharsetUtil.ISO_8859_1));
        buf.release();
    }
}

/**
 * Start HBase connection
 *
 * @param ch channel to start connection on
 */
private void startHBaseConnection(Channel ch) {
  ch.pipeline()
      .addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
  ch.pipeline().addLast(new AsyncServerResponseHandler(this));

  try {
    writeChannelHeader(ch).addListener(new GenericFutureListener<ChannelFuture>() {
      @Override public void operationComplete(ChannelFuture future) throws Exception {
        if (!future.isSuccess()) {
          close(future.cause());
          return;
        }
        for (AsyncCall call : calls.values()) {
          writeRequest(call);
        }
      }
    });
  } catch (IOException e) {
    close(e);
  }
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
  LOGGER.info("Setting up Server channel, scheduler");
  ch.pipeline().addLast("decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
  ch.pipeline().addLast("encoder", new LengthFieldPrepender(4));
  //ch.pipeline().addLast("logger", new LoggingHandler());
  // Create server metric for this handler and add to aggregate if present
  NettyServerMetrics serverMetric =
      new NettyServerMetrics(_registry, NettyTCPServer.class.getName() + "_" + Utils.getUniqueId() + "_");

  if (null != _globalMetrics) {
    _globalMetrics.addTransportClientMetrics(serverMetric);
  }

  ch.pipeline().addLast("request_handler",
      new NettyChannelInboundHandler(_handlerFactory.createNewRequestHandler(), serverMetric, _defaultLargeQueryLatencyMs));
}

private static <T> void serveMany(int port, final Observable<List<Observable<T>>> observable, final Encoder<T> encoder,
            boolean startAndWait, SlottingStrategy<T> slottingStrategy, IngressPolicy ingressPolicy){
        RxServer<RemoteRxEvent, RemoteRxEvent> server 
            = RxNetty.createTcpServer(port, new PipelineConfiguratorComposite<RemoteRxEvent, RemoteRxEvent>(
                new PipelineConfigurator<RemoteRxEvent, RemoteRxEvent>(){
                    @Override
                    public void configureNewPipeline(ChannelPipeline pipeline) {
//                      pipeline.addFirst(new LoggingHandler(LogLevel.ERROR)); // uncomment to enable debug logging 
                        pipeline.addLast("frameEncoder", new LengthFieldPrepender(4)); // 4 bytes to encode length
                        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(524288, 0, 4, 0, 4)); // max frame = half MB
                    }
                }, new RxEventPipelineConfigurator()),  
                new RemoteObservableConnectionHandler<T>(observable, encoder, slottingStrategy, ingressPolicy));
        if(startAndWait){
            server.startAndWait();
        }else{
            server.start();
        }
    }

/**
 * Returns a new channel initializer suited to encode and decode a protocol
 * buffer message.
 * <p/>
 * <p>Message sizes over 10 MB are not supported.</p>
 * <p/>
 * <p>The handler will be executed on the I/O thread. Blocking operations
 * should be executed in their own thread.</p>
 *
 * @param defaultInstance an instance of the message to handle
 * @param handler the handler implementing the application logic
 * @param <M> the type of the support protocol buffer message
 */
public static final <M extends Message> ChannelInitializer<Channel> protoBuf(
    final M defaultInstance, final SimpleChannelInboundHandler<M> handler) {
  return new ChannelInitializer<Channel>() {

    @Override
    protected void initChannel(Channel channel) throws Exception {
      channel.pipeline().addLast("frameDecoder",
          new LengthFieldBasedFrameDecoder(10 * 1024 * 1024, 0, 4, 0, 4));
      channel.pipeline().addLast("protobufDecoder",
          new ProtobufDecoder(defaultInstance));
      channel.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
      channel.pipeline().addLast("protobufEncoder", new ProtobufEncoder());
      channel.pipeline().addLast("applicationHandler", handler);
    }
  };
}

@Override
protected void initChannel(SocketChannel channel) throws Exception {
    channel.pipeline()
            .addLast(new ReadTimeoutHandler(30))
            .addLast("splitter", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4))
            .addLast(new PacketDecoder())
            .addLast("prepender", new LengthFieldPrepender(4))
            .addLast(new PacketEncoder())
            .addLast(client.getHandler());
    this.client.setChannel(channel);
    System.out.println("Netty client started");
}

public void start() {
  Configuration conf = context.getConf();
  int workerNum =
      conf.getInt(AngelConf.ANGEL_NETTY_MATRIXTRANSFER_SERVER_EVENTGROUP_THREADNUM,
          AngelConf.DEFAULT_ANGEL_NETTY_MATRIXTRANSFER_SERVER_EVENTGROUP_THREADNUM);

  int sendBuffSize =
      conf.getInt(AngelConf.ANGEL_NETTY_MATRIXTRANSFER_SERVER_SNDBUF,
          AngelConf.DEFAULT_ANGEL_NETTY_MATRIXTRANSFER_SERVER_SNDBUF);

  int recvBuffSize =
      conf.getInt(AngelConf.ANGEL_NETTY_MATRIXTRANSFER_SERVER_RCVBUF,
          AngelConf.DEFAULT_ANGEL_NETTY_MATRIXTRANSFER_SERVER_RCVBUF);

  final int maxMessageSize =
      conf.getInt(AngelConf.ANGEL_NETTY_MATRIXTRANSFER_MAX_MESSAGE_SIZE,
          AngelConf.DEFAULT_ANGEL_NETTY_MATRIXTRANSFER_MAX_MESSAGE_SIZE);

  bossGroup = new NioEventLoopGroup(1);
  workerGroup = new NioEventLoopGroup(workerNum);
  ((NioEventLoopGroup) workerGroup).setIoRatio(70);

  ServerBootstrap b = new ServerBootstrap();
  b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
      .option(ChannelOption.SO_SNDBUF, sendBuffSize)
      .option(ChannelOption.SO_RCVBUF, recvBuffSize)
      .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
          ChannelPipeline p = ch.pipeline();
          p.addLast(new LengthFieldBasedFrameDecoder(maxMessageSize, 0, 4, 0, 4));
          p.addLast(new LengthFieldPrepender(4));
          p.addLast(new MatrixTransportServerHandler(context));
        }
      });

  channelFuture = b.bind(port);
}

/**
 * Creates a LengthFieldBasedFrameDecoder where the first 8 bytes are the length of the frame.
 * This is used before all decoders.
 */
public static ByteToMessageDecoder createFrameDecoder() {
  // maxFrameLength = 2G
  // lengthFieldOffset = 0
  // lengthFieldLength = 8
  // lengthAdjustment = -8, i.e. exclude the 8 byte length itself
  // initialBytesToStrip = 8, i.e. strip out the length field itself
  return new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 8, -8, 8);
}

@Override
protected void initChannel(T ch) throws Exception {
    ChannelPipeline p = ch.pipeline();

    p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
    p.addLast("protobufDecoder", new ProtobufDecoder(NettyRpcProto.RpcContainer.getDefaultInstance()));

    p.addLast("frameEncoder", new LengthFieldPrepender(4));
    p.addLast("protobufEncoder", new ProtobufEncoder());

    ConcurrentHashMap<Integer, RpcCall> callMap = new ConcurrentHashMap<Integer, RpcCall>();
    p.addLast(eventExecutor, "inboundHandler", new InboundHandler(callMap));
    p.addLast("outboundHandler", new OutboundHandler(callMap));

}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline p = ch.pipeline();

    p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
    p.addLast("protobufDecoder", new ProtobufDecoder(NettyRpcProto.RpcContainer.getDefaultInstance()));

    p.addLast("frameEncoder", new LengthFieldPrepender(4));
    p.addLast("protobufEncoder", new ProtobufEncoder());

    p.addLast(eventExecutor, "serverHandler", handler);
}

@Override
public void run() {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                        pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                        pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
                        pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
                        pipeline.addLast(new SimpleClientChannelHandler());
                    }

                });
        ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
        if (channelFuture.isSuccess()) {
            System.out.println(String.format("connect server(%s:%s) sucess", host, port));
        }
        channelFuture.channel().closeFuture().sync();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        group.shutdownGracefully();
    }
}

@Override
public void start() {
    int heartTime = transaction_netty_heart_time+10;
    txCoreServerHandler = new TxCoreServerHandler(mqTxManagerService);
    bossGroup = new NioEventLoopGroup(50); // (1)
    workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 100)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast("timeout", new IdleStateHandler(heartTime, heartTime, heartTime, TimeUnit.SECONDS));

                        ch.pipeline().addLast(new LengthFieldPrepender(4, false));
                        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));

                        ch.pipeline().addLast(txCoreServerHandler);
                    }
                });

        // Start the server.
        b.bind(Constants.socketPort);
        logger.info("Socket started on port(s): " + Constants.socketPort + " (socket)");

    } catch (Exception e) {
        // Shut down all event loops to terminate all threads.
        e.printStackTrace();
    }
}

private ChannelInitializer channelInitializer() {
    return new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            ChannelPipeline pipeline = socketChannel.pipeline();
            pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
            pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));
            pipeline.addLast("handler", nodeChannelInBoundHandler());
        }
    };
}

@PostConstruct
public void start() throws Exception {

    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workGroup)
            .channelFactory(NioServerSocketChannel::new)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {

                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                    pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
                    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                    pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
                    pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));
                    pipeline.addLast("handler", nodeChannelInBoundHandler);
                }
            })
            .option(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_SNDBUF, 2048)
            .option(ChannelOption.SO_RCVBUF, 1024);
    bootstrap.bind(goPushNodeServerConfig.getNodePort()).sync();
    log.info("Node server start successful! listening port: {}", goPushNodeServerConfig.getNodePort());
}