@Test
public void serverBootStrapWithOptionstest() throws InstantiationException,illegalaccessexception,ClassNotFoundException {
    LinkedHashMap<String,Object> channelHandlerOptions = new LinkedHashMap<String,Object>();

    channelHandlerOptions.put("lineFrame",new LineBasedFrameDecoder(2000));
    channelHandlerOptions.put("decoder",new StringDecoder());
    channelHandlerOptions.put("encoder",new StringEncoder());
    channelHandlerOptions.put("handler",new ChannelInboundHandlerAdapter() {

        @Override
        public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception {
            log.info("Message Received and forward to ConsumerProcessor. Msg -> {}",msg);
        }
    });

    Server server = BootStrap.builder()
            .port(5252)
            .options(channelHandlerOptions)
            .messageConsumer(msg -> log.info(msg))
            .build();

    assertNotNull(server);
}

public void connect(int port,String host) throws InterruptedException {

    EventLoopGroup group = new NioEventLoopGroup();
    try{
        Bootstrap b = new Bootstrap();
        b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true)
        .handler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                ch.pipeline().addLast(new StringDecoder());
                ch.pipeline().addLast(new TimeClientHandler());
            }

        });

        //发送异步链接操作
        ChannelFuture f = b.connect(host,port).sync();
        //等待客户端链路关闭
        f.channel().closeFuture().sync();
    }finally{
        group.shutdownGracefully();
    }

}

@Test
public void addByteDecoderWhenNoLeft() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.ReactiveBridge,new ChannelHandlerAdapter() {
           });
    ChannelHandler decoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerLast("decoder",decoder)
               .addHandlerFirst("decoder$extract",NettyPipeline.inboundHandler(ADD_EXTRACTOR));

    assertEquals(channel.pipeline()
                        .names(),Arrays.asList("decoder$extract","decoder",NettyPipeline.ReactiveBridge,"DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteDecoderWhennoright() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec,Arrays.asList(NettyPipeline.HttpCodec,"decoder$extract","DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteDecoderWhenFullReactorPipeline() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec,new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler,new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge,NettyPipeline.HttpServerHandler,"DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteEncoderWhenNoLeft() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.ReactiveBridge,new ChannelHandlerAdapter() {
           });
    ChannelHandler encoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerFirst("encoder",encoder);

    assertEquals(channel.pipeline()
                        .names(),Arrays.asList("encoder","DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteEncoderWhennoright() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec,"encoder","DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteEncoderWhenFullReactorPipeline() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec,"DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addSeveralByteEncodersWhenCodec() throws Exception {
    ChannelHandler encoder1 = new LineBasedFrameDecoder(12);
    ChannelHandler encoder2 = new LineBasedFrameDecoder(13);

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec,new ChannelHandlerAdapter() {
           });

    testContext.addHandlerFirst("encoder1",encoder1)
               .addHandlerFirst("encoder2",encoder2);

    assertEquals(channel.pipeline()
                        .names(),"encoder2","encoder1","DefaultChannelPipeline$TailContext#0"));
}

@Test
public void flushOnComplete() {

    Flux<String> test = Flux.range(0,100)
                            .map(n -> String.format("%010d",n));

    NettyContext c = HttpServer.create(0)
                               .newHandler((req,resp) -> resp.sendString(test.map(s -> s + "\n")))
                               .block(Duration.ofSeconds(30));

    Flux<String> client = HttpClient.create(c.address()
                                             .getPort())
                                    .get("/")
                                    .block(Duration.ofSeconds(30))
                                    .addHandler(new LineBasedFrameDecoder(10))
                                    .receive()
                                    .asstring();

    StepVerifier.create(client)
                .expectNextSequence(test.toIterable())
                .expectComplete()
                .verify(Duration.ofSeconds(30));

    c.dispose();
}

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(RxtxChannel.class)
         .handler(new ChannelInitializer<RxtxChannel>() {
             @Override
             public void initChannel(RxtxChannel ch) throws Exception {
                 ch.pipeline().addLast(
                     new LineBasedFrameDecoder(32768),new StringEncoder(),new StringDecoder(),new RxtxClientHandler()
                 );
             }
         });

        ChannelFuture f = b.connect(new RxtxDeviceAddress(PORT)).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(RxtxChannel.class)
         .handler(new ChannelInitializer<RxtxChannel>() {
             @Override
             public void initChannel(RxtxChannel ch) throws Exception {
                 ch.pipeline().addLast(
                     new LineBasedFrameDecoder(32768),new RxtxClientHandler()
                 );
             }
         });

        ChannelFuture f = b.connect(new RxtxDeviceAddress(PORT)).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    LineBasedFrameDecoder lineDecoder = new LineBasedFrameDecoder(MAX_LINE_LENGTH);
    StringDecoder stringDecoder = new StringDecoder(CHARSET); //FIXME: Should only split on CRLF,not on LF alone
    MessageDecoder messageDecoder = new MessageDecoder();
    MessageHandler messageHandler = new MessageHandler(handler);

    StringEncoder stringEncoder = new StringEncoder(CHARSET);
    MessageEncoder messageEncoder = new MessageEncoder();

    IdleStateHandler idleHandler = new IdleStateHandler(IDLE_TIMEOUT,0);

    // Inbound goes from first to last,outbound goes from last to first.
    // i.e. the outside is on the left/top,the inside is on the right/bottom
    ch.pipeline().addLast(lineDecoder).addLast(stringDecoder).addLast(messageDecoder).addLast(idleHandler).addLast(messageHandler)
            .addLast(stringEncoder).addLast(messageEncoder);

}

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(RxtxChannel.class)
                .handler(new ChannelInitializer<RxtxChannel>() {
                    @Override
                    public void initChannel(RxtxChannel ch) throws Exception {
                        ch.pipeline().addLast(
                                new LineBasedFrameDecoder(32768),new RxtxClientHandler()
                        );
                    }
                });

        ChannelFuture f = b.connect(new RxtxDeviceAddress(PORT)).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

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 LineBasedFrameDecoder(20))
                                    .addLast(new StringDecoder())
                                    .addLast(new DelimitedClientHandler());
                        }
                    });

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

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(RxtxChannel.class)
         .handler(new ChannelInitializer<RxtxChannel>() {
             @Override
             public void initChannel(RxtxChannel ch) throws Exception {
                 ch.pipeline().addLast(
                     new LineBasedFrameDecoder(32768),new RxtxClientHandler()
                 );
             }
         });

        ChannelFuture f = b.connect(new RxtxDeviceAddress("/dev/ttyUSB0")).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

public static void main(String[] args) throws Exception {
    EventLoopGroup group = new OioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(RxtxChannel.class)
         .handler(new ChannelInitializer<RxtxChannel>() {
             @Override
             public void initChannel(RxtxChannel ch) throws Exception {
                 ch.pipeline().addLast(
                     new LineBasedFrameDecoder(32768),new RxtxClientHandler()
                 );
             }
         });

        ChannelFuture f = b.connect(new RxtxDeviceAddress("/dev/ttyUSB0")).sync();

        f.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully();
    }
}

public SimpleLineBasedSerialChannel(String port,final SimpleStringChannelHandler stringHandler) {
    group = new OioEventLoopGroup();
       Bootstrap b = new Bootstrap();
       b.group(group)
        .channel(JsscChannel.class)
        .handler(new ChannelInitializer<JsscChannel>() {
            @Override
            public void initChannel(JsscChannel ch) throws Exception {
                ch.pipeline().addLast(
                    new LineBasedFrameDecoder(Integer.MAX_VALUE),new SimpleChannelInboundHandler<String>() {
                     @Override
                     protected void channelRead0(io.netty.channel.ChannelHandlerContext ctx,String msg) throws Exception {
                        stringHandler.channelRead(ctx,msg); 
                     }
                 }
                );
            }
        });

        f = b.connect(new JsscDeviceAddress(port)).syncUninterruptibly();
}

@Test
public void testDecode() throws IOException {
    EmbeddedChannel decoderEmbedder = new EmbeddedChannel(new LineBasedFrameDecoder(4096),new YencDecoder());

    ByteBuf encoded = Unpooled.buffer();
    IoUtils.copy(Resources.getResource("lorem-ipsum.ync").openStream(),new ByteBufOutputStream(encoded));

    ByteBuf original = Unpooled.buffer();
    IoUtils.copy(Resources.getResource("lorem-ipsum").openStream(),new ByteBufOutputStream(original));

    decoderEmbedder.writeInbound(encoded);
    decoderEmbedder.finish();

    Object[] result = decoderEmbedder.inboundMessages().toArray();
    ByteBuf[] buffers = Arrays.copyOf(result,result.length,ByteBuf[].class);
    assertthat(Unpooled.copiedBuffer(buffers),exactChannelBuffer(original));
}

@Test(expected = YencChecksumFailureException.class)
public void testChecksumFailure() throws Throwable {
    EmbeddedChannel decoderEmbedder = new EmbeddedChannel(new LineBasedFrameDecoder(4096),new YencDecoder());

    ByteBuf encoded = Unpooled.buffer();
    IoUtils.copy(Resources.getResource("lorem-ipsum-invalid-checksum.ync").openStream(),new ByteBufOutputStream(original));

    try {
        decoderEmbedder.writeInbound(encoded);
        decoderEmbedder.finish();
    }
    catch(DecoderException cee) {
        throw cee.getCause();
    }
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    //增加以\n 和 \r\n为数据换行符的Handler
    ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
    //增加字符串解析器
    ch.pipeline().addLast(new StringDecoder());
    //对输入数据进行业务逻辑处理
    ch.pipeline().addLast(new RightTimeServerHandler());
}

/**
 *@description 连接服务器
 *@time 创建时间:2017年7月21日下午4:15:50
 *@param host
 *@param port
 *@throws InterruptedException
 *@author dzn
 */
public void connect(String host,int port) throws InterruptedException{
    EventLoopGroup group = new NioEventLoopGroup();
    try{
        Bootstrap boot = new Bootstrap();
        boot.group(group)
        .channel(NioSocketChannel.class)
        .option(ChannelOption.TCP_NODELAY,true)
        .handler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                //增加以\n 和 \r\n为数据换行符的Handler
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                //增加字符串解析器
                ch.pipeline().addLast(new StringDecoder());
                //对输入数据进行业务逻辑处理
                ch.pipeline().addLast(new RightTimeClientHandler());
            }

        });

        //连接服务器
        ChannelFuture future = boot.connect(host,port).sync();

        //等待客户端Channel关闭
        future.channel().closeFuture().sync();

    }finally{
        group.shutdownGracefully();
    }
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ch.pipeline().addLast(new IdleStateHandler(5,5,10));
    ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
    ch.pipeline().addLast(new StringDecoder());
    ch.pipeline().addLast(new ServerHandler());
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ch.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
    ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
    ch.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
    ch.pipeline().addLast(new FileServerHandler());
}

public void bind(int port) throws Exception {

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup,workerGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.so_BACKLOG,1024)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {

                    ChannelPipeline p = ch.pipeline();
                    p.addLast(new LineBasedFrameDecoder(1024));
                    p.addLast(new StringDecoder());
                    p.addLast(new StringEncoder());

                    p.addLast(new LineserverHandler());
                }
            });

            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(port).sync(); // (7)

            logger.info("server bind port:{}",port);

            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();

        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

public void connect(String host,int port) throws InterruptedException {

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY,true)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {

                    ChannelPipeline p = ch.pipeline();
                    p.addLast(new LineBasedFrameDecoder(1024));
                    p.addLast(new StringDecoder());
                    p.addLast(new StringEncoder());

                    p.addLast(new LineClientHandler());
                }
            });

            ChannelFuture future = b.connect(Constants.HOST,Constants.PORT).sync();

            future.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

/**
 * 채널 파이프라인 설정.
 * Netty.Server.Configuration.NettyServerConfiguration 에서 등록한 Bean 을 이용해 사용자의 통신을 처리할 Handler 도 등록.
 * Netty.Server.Handler.JsonHandler 에서 실제 사용자 요청 처리.
 *
 * @param channel
 * @throws Exception
 */
@Override
protected void initChannel(Channel channel) throws Exception {

    ChannelPipeline channelPipeline = channel.pipeline();

    switch (transferType) {

        case "websocket":

            channelPipeline
                    .addLast(new HttpServerCodec())
                    .addLast(new HttpObjectAggregator(65536))
                    .addLast(new WebSocketServerCompressionHandler())
                    .addLast(new WebSocketServerProtocolHandler(transferWebsocketPath,transferWebsocketSubProtocol,transferWebsocketAllowExtensions))
                    .addLast(new LoggingHandler(LogLevel.valueOf(logLevelPipeline)))
                    .addLast(websocketHandler);

        case "tcp":
        default:

            channelPipeline
                    .addLast(new LineBasedFrameDecoder(Integer.MAX_VALUE))
                    .addLast(STRING_DECODER)
                    .addLast(STRING_ENCODER)
                    .addLast(new LoggingHandler(LogLevel.valueOf(logLevelPipeline)))
                    .addLast(jsonHandler);

    }

}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
     // 以下两行代码为了解决半包读问题  
       ch.pipeline().addLast(new LineBasedFrameDecoder(1024));  
       ch.pipeline().addLast(new StringDecoder());  

       ch.pipeline().addLast(new TimeServerHandler()); 
}

public void connect(int port,String host) {
    // 配置客户端的NIO线程组
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch)
                            throws Exception {
                        // 以下两行代码为了解决半包读问题
                        ch.pipeline().addLast(
                                new LineBasedFrameDecoder(1024));
                        ch.pipeline().addLast(new StringDecoder());

                        ch.pipeline().addLast(new TimeClientHandler());
                    }
                });
        // 发起异步连接操作
        ChannelFuture f = b.connect(host,port).sync();

        // 等待链路关闭
        f.channel().closeFuture().sync();

    } catch (Exception e) {
    } finally {
        // 退出，释放NIO线程组
        group.shutdownGracefully();
    }
}

/**
 * {@inheritDoc}
 * @see io.netty.channel.ChannelInitializer#initChannel(io.netty.channel.Channel)
 */
@Override
protected void initChannel(final AbstractChannel ch) throws Exception {
    ChannelPipeline p = ch.pipeline();
    p.addLast("bytesDecoder",bytesDecoder);
    p.addLast("framer",new LineBasedFrameDecoder(1024,true,true));
    p.addLast("linehandler",new StringMetricHandler());
}

@Test
public void addByteDecoderWhenEmptyPipeline() throws Exception {

    ChannelHandler decoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerLast("decoder","DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addSeveralByteDecodersWhenCodec() throws Exception {
    ChannelHandler decoder1 = new LineBasedFrameDecoder(12);
    ChannelHandler decoder2 = new LineBasedFrameDecoder(13);

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec,new ChannelHandlerAdapter() {
           });

    testContext.addHandlerLast("decoder1$extract",NettyPipeline.inboundHandler(ADD_EXTRACTOR))
               .addHandlerLast("decoder1",decoder1)

               .addHandlerLast("decoder2$extract",NettyPipeline.inboundHandler(ADD_EXTRACTOR))
               .addHandlerLast("decoder2",decoder2);

    assertEquals(channel.pipeline()
                        .names(),"decoder1$extract","decoder1","decoder2$extract","decoder2","DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addByteEncoderWhenEmptyPipeline() throws Exception {

    ChannelHandler encoder = new LineBasedFrameDecoder(12);

    testContext.addHandlerFirst("encoder","DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addDecoderSkipsIfExist() {
    channel.pipeline()
           .addFirst("foo",new Utf8FrameValidator());

    testContext.addHandlerFirst("foo",new LineBasedFrameDecoder(10));

    assertEquals(channel.pipeline()
                        .names(),Arrays.asList("foo","DefaultChannelPipeline$TailContext#0"));
    assertthat(channel.pipeline()
                      .get("foo"),is(instanceOf(Utf8FrameValidator.class)));
}

@Test
public void addEncoderSkipsIfExist() {
    channel.pipeline()
           .addFirst("foo",is(instanceOf(Utf8FrameValidator.class)));
}

@Test
public void tcpClientHandlesLineFeedData() throws InterruptedException {
    final int messages = 100;
    final CountDownLatch latch = new CountDownLatch(messages);
    final List<String> strings = new ArrayList<String>();

            TcpClient.create(opts -> opts.host("localhost")
                                         .port(echoServerPort)
                                         .afterChannelInit(c -> c.pipeline()
                                                                 .addBefore(
                                                                         NettyPipeline.ReactiveBridge,"codec",new LineBasedFrameDecoder(
                                                                                 8 * 1024))))
                     .newHandler((in,out) ->
                        out.sendString(Flux.range(1,messages)
                                            .map(i -> "Hello World!" + i + "\n")
                                            .subscribeOn(Schedulers.parallel()))
                            .then( in.receive()
                                     .asstring()
                                     .take(100)
                                     .flatMapIterable(s -> Arrays.asList(s.split("\\n")))
                                     .doOnNext(s -> {
                                         strings.add(s);
                                         latch.countDown();
                                     }).then())
                     )
                     .block(Duration.ofSeconds(15))
                     .onClose()
                     .block(Duration.ofSeconds(30));

    assertTrue("Expected messages not received. Received " + strings.size() + " messages: " + strings,latch.await(15,TimeUnit.SECONDS));

    assertEquals(messages,strings.size());
}

@Provides
@WhoisProtocol
static ImmutableList<Provider<? extends ChannelHandler>> provideHandlerProviders(
    @WhoisProtocol Provider<ReadTimeoutHandler> readTimeoutHandlerProvider,Provider<LineBasedFrameDecoder> lineBasedFrameDecoderProvider,Provider<WhoisServiceHandler> whoisServiceHandlerProvider,Provider<LoggingHandler> loggingHandlerProvider,Provider<FullHttpRequestRelayHandler> relayHandlerProvider) {
  return ImmutableList.of(
      readTimeoutHandlerProvider,lineBasedFrameDecoderProvider,whoisServiceHandlerProvider,loggingHandlerProvider,relayHandlerProvider);
}

@Override
public ChannelInitializer<Channel> initializer() {
    return new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(final Channel ch) throws Exception {
            ch.pipeline().addLast(new LineBasedFrameDecoder(MAX_LINE));
            ch.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
            ch.pipeline().addLast(decoder,handler);
        }
    };
}

@Override
public final ChannelInitializer<Channel> initializer() {
    return new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            ch.pipeline().addLast(new LineBasedFrameDecoder(MAX_LINE));
            ch.pipeline().addLast(decoder,handler);
        }
    };
}

public void run(int port) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
    ServerBootstrap b = new ServerBootstrap();
    b.group(bossGroup,workerGroup)
        .channel(NioServerSocketChannel.class)
        .option(ChannelOption.so_BACKLOG,100)
        .childHandler(new ChannelInitializer<SocketChannel>() {
        /*
         * (non-Javadoc)
         * 
         * @see
         * io.netty.channel.ChannelInitializer#initChannel(io
         * .netty.channel.Channel)
         */
        public void initChannel(SocketChannel ch)
            throws Exception {
            ch.pipeline().addLast(
                new StringEncoder(CharsetUtil.UTF_8),new LineBasedFrameDecoder(1024),new StringDecoder(CharsetUtil.UTF_8),new FileServerHandler());
        }
        });
    ChannelFuture f = b.bind(port).sync();
    System.out.println("Start file server at port : " + port);
    f.channel().closeFuture().sync();
} finally {
    // 优雅停机
    bossGroup.shutdownGracefully();
    workerGroup.shutdownGracefully();
}
   }

public void start() throws Exception {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(this.host,this.port))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        System.out.println("connected server...");
                        ch.pipeline().addLast(new ByteArrayEncoder());
                        ch.pipeline().addLast(new ByteArrayDecoder());
                        ch.pipeline().addLast(new EchoClientHandler());
                    }
                });

        ChannelFuture cf = b.connect().sync();

        cf.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully().sync();
    }
}

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

  final UUID uuid = UUID.randomUUID();

  LOG.debug("KaaTcpserverInitializer Initializing Channel {} connection from {}:{}",uuid,ch.remoteAddress().getAddress().toString(),ch.remoteAddress().getPort());

  Attribute<UUID> uuidAttr = ch.attr(AbstractNettyServer.UUID_KEY);
  uuidAttr.set(uuid);

  p.addLast("binaryDecoder",new ByteArrayDecoder());
  p.addLast("kaaTcpDecoder",getDecoder());
  p.addLast("binaryEncoder",new ByteArrayEncoder());
  p.addLast("kaaTcpEncoder",new KaaTcpEncoder());
  p.addLast("mainHandler",getMainHandler(uuid));
  p.addLast("kaaTcpExceptionHandler",new KaaTcpExceptionHandler());
}

@Override
public synchronized void start() {
    bossGroup = new NioEventLoopGroup(); // (1)
    workerGroup = new NioEventLoopGroup();
    try {
        b = new ServerBootstrap(); // (2)
        b.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class) // (3)
                .childHandler(new ChannelInitializer<SocketChannel>() { // (4)
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {

                        ch.pipeline().addLast(new ByteArrayDecoder());
                        ch.pipeline().addLast(new ByteArrayEncoder());

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

                        ch.pipeline().addLast(new IdleStateHandler(heartTime,heartTime,TimeUnit.SECONDS));

                        ch.pipeline().addLast(new DeliveryHandler(deliveryService));

                    }
                })
                .option(ChannelOption.so_BACKLOG,128)          // (5)
                .childOption(ChannelOption.so_KEEPALIVE,true); // (6)

        // Bind and start to accept incoming connections.
        b.bind(settingService.getDeliveryPort());

        logger.info("socket: "+settingService.getDeliveryPort()+" starting....");
        // Wait until the server socket is closed.
        // In this example,this does not happen,but you can do that to gracefully
    } catch (Exception e) {
        e.printstacktrace();
    }
}

public static ChannelHandlerFactory newByteArrayDecoder(String protocol) {
    if ("udp".equals(protocol)) {
        return new ShareableChannelHandlerFactory(new DatagramPacketByteArrayDecoder());
    } else {
        return new ShareableChannelHandlerFactory(new ByteArrayDecoder());
    }
}

public ChannelInitializerImpl(int command,String message) {
    this.handler_list = new ChannelHandler[]{new CMDFieldPrepender(command),new LengthFieldPrepender(4),//new StringEncoder(CommonConstants.UTF8),new StringDecoder(CommonConstants.UTF8),new StringEncoder(CommonConstants.UTF8),new ByteArrayDecoder(),//new MessageClientHandler(message)};
        new ByteMessageClientHandler(message)};
        //new DelimiterBasedFrameDecoder(2048,Delimiters.lineDelimiter()),}

public void open(EventLoopGroup eventLoopGroup) throws Exception {
    if (openned.compareAndSet(false,true)) {
        eventloopGroop = eventLoopGroup == null ? new NioEventLoopGroup()
                : eventLoopGroup;
        Bootstrap bootstrap = new Bootstrap();
        final BlockingByteArrayClientHandler handler = new BlockingByteArrayClientHandler(
                this);
        this.clientHandler = handler;
        bootstrap.group(eventloopGroop).channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch)
                            throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        SSLEngine engine = SecureSocketSslContextFactory
                                .getClientContext().createSSLEngine();
                        engine.setUseClientMode(true);
                        pipeline.addLast("ssl",new SslHandler(engine));
                        pipeline.addLast("length-decoder",new LengthFieldBasedFrameDecoder(
                                        Integer.MAX_VALUE,4));
                        pipeline.addLast("bytearray-decoder",new ByteArrayDecoder());
                        pipeline.addLast("length-encoder",new LengthFieldPrepender(4));
                        pipeline.addLast("bytearray-encoder",new ByteArrayEncoder());
                        pipeline.addLast("handler",handler);
                    }

                });
        channelFuture = bootstrap.connect(this.remoteHost,this.remotePort)
                .sync();
    }
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    SSLEngine engine =
        SecureSocketSslContextFactory.getServerContext().createSSLEngine();
    engine.setUseClientMode(false);
    pipeline.addLast("ssl",new SslHandler(engine));  
    pipeline.addLast("length-decoder",new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,4)); 
    pipeline.addLast("bytearray-decoder",new ByteArrayDecoder());
    pipeline.addLast("length-encoder",new LengthFieldPrepender(4));  
    pipeline.addLast("bytearray-encoder",new ByteArrayEncoder());
    pipeline.addLast("handler",new SecureSocketServerhandler2());
}

public GossipServerThread() {
    gossipBossGroup = new NioEventLoopGroup(GOSSIP_BOSS_THREADS);
    gossipworkerGroup = new NioEventLoopGroup(GOSSIP_WORKER_THREADS);
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(gossipBossGroup,gossipworkerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.config().setAllocator(UnpooledByteBufAllocator.DEFAULT);
                        ch.pipeline().addLast(new GossipMessageDecoder());
                        ch.pipeline().addLast("encoder",new ByteArrayEncoder());
                        ch.pipeline().addLast("decoder",new ByteArrayDecoder());
                        ch.pipeline().addLast(new GossipMessageHandler());
                        if(LOG.isTraceEnabled()) {
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.TRACE));
                        }
                    }

                    @Override
                    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) throws Exception {
                        LOG.error("Cannot initialize gossip server.",cause);
                    }
                })
                .option(ChannelOption.so_BACKLOG,128)
                .childOption(ChannelOption.so_KEEPALIVE,true);

        // Bind and start to accept incoming connections.
        f = b.bind(getIp(),getGossipPort()).sync();
    } catch (InterruptedException ex) {
        LOG.error("Gossip server interrupted.",ex);
    }
}

public DataServerThread() {
    dataBossGroup = new NioEventLoopGroup(DATA_BOSS_THREADS);
    dataWorkerGroup = new NioEventLoopGroup(DATA_WORKER_THREADS);
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(dataBossGroup,dataWorkerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.config().setAllocator(UnpooledByteBufAllocator.DEFAULT);
                        if(useSSL) {
                            ch.pipeline().addLast(sslContext.newHandler(ch.alloc()));
                        }
                        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(32768,2,2));
                        ch.pipeline().addLast("decoder",new ByteArrayDecoder());
                        ch.pipeline().addLast(new DataMessageHandler());
                        if(LOG.isTraceEnabled()) {
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.TRACE));
                        }
                    }

                    @Override
                    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) throws Exception {
                        LOG.error("Cannot initialize data server.",cause);
                    }
                })
                .option(ChannelOption.so_SNDBUF,262144)
                .option(ChannelOption.so_RCVBUF,262144)
                .option(ChannelOption.so_BACKLOG,getDataPort()).sync();
    } catch (InterruptedException ex) {
        LOG.error("Message data interrupted.",ex);
    }
}

public GossipServerThread() {
    gossipBossGroup = new NioEventLoopGroup(GOSSIP_BOSS_THREADS);
    gossipworkerGroup = new NioEventLoopGroup(GOSSIP_WORKER_THREADS);
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(gossipBossGroup,new ByteArrayDecoder());
                        ch.pipeline().addLast(new GossipMessageHandler());
                        if (LOG.isTraceEnabled()) {
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.TRACE));
                        }
                    }

                    @Override
                    public void exceptionCaught(ChannelHandlerContext ctx,ex);
    }
}

@Override
protected void initChannel(final SocketChannel ch) throws Exception {
  ch.pipeline()
      .addLast("frameDecoder",new LengthFieldBasedFrameDecoder(MAXFRAMELENGTH,4))
      .addLast("bytesDecoder",new ByteArrayDecoder())
      .addLast("frameEncoder",new LengthFieldPrepender(4))
      .addLast("bytesEncoder",new ByteArrayEncoder())
      .addLast("chunker",new ChunkedReadWriteHandler())
      .addLast("handler",handlerFactory.createChannelInboundHandler());
}

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

    // add the IP ACL filter first
    if (ipFilter != null) {
        p.addLast("acl",ipFilter);
    }

    if (sslCtx != null) {
        if (configuration.isClientAuth()) {
            final SSLEngine engine = sslCtx.newEngine(ch.alloc());
            engine.setUseClientMode(false);
            engine.setNeedClientAuth(true);

            p.addLast("ssl",new SslHandler(engine));
        } else {
            p.addLast("ssl",sslCtx.newHandler(ch.alloc()));
        }
    }

    // removes idle connections after READER_IDLE_SECONDS seconds
    p.addLast("idleStateHandler",new IdleStateHandler(READER_IDLE_SECONDS,0));

    // authenticate via an ACL and mutual certificates
    p.addLast("auth",new AuthHandler(configuration.isClientAuth()));

    // check to see if the data stream is gzipped or not
    // p.addLast("gzipDetector",new OptionalGzipHandler());

    // break each data chunk by newlines
    p.addLast("line",new LineBasedFrameDecoder(Ints.checkedCast(maxLength),true,true));

    // convert each data chunk into a byte array
    p.addLast("decoder",new ByteArrayDecoder());

    // batch and compress chunks of data up to maxUploadBytes
    p.addLast("batcher",new BatchHandler(maxUploadBytes));

    // upload the batch to S3
    p.addLast("uploader",uploadHandler);
}

private void initializeGossipClient() {
    LOG.trace("Initializing gossip client");

    gossipworkerGroup = new NioEventLoopGroup(GOSSIP_WORKER_THREADS);

    Bootstrap b = new Bootstrap();
    b.group(gossipworkerGroup);
    b.channel(NioSocketChannel.class);
    b.option(ChannelOption.so_KEEPALIVE,true);
    b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,timeout);
    b.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ch.config().setAllocator(UnpooledByteBufAllocator.DEFAULT);
            ch.pipeline().addLast("encoder",new ByteArrayEncoder());
            ch.pipeline().addLast("decoder",new ByteArrayDecoder());
            ch.pipeline().addLast(new GossipExceptionHandler());
            if(LOG.isTraceEnabled()) {
                ch.pipeline().addLast(new LoggingHandler(LogLevel.TRACE));
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) throws Exception {
            LOG.error("Cannot initialize gossip client.",cause);
            ctx.close();
        }
    });

    // Start the client.
    ChannelFuture future = b.connect(getIp(),getGossipPort()).awaitUninterruptibly();

    if(future.isCancelled()) {
        gossipChannel = null;
    } else if(!future.isSuccess()) {
        gossipChannel = null;
        retryGossipConnection();
    } else {
        gossipChannel = future.channel();
        setStatus(MemberStatus.Alive);
        updateMember();
    }
}

private void initializeDataClient() {
    LOG.trace("Initializing data client");

    dataWorkerGroup = new NioEventLoopGroup(DATA_WORKER_THREADS);

    Bootstrap b = new Bootstrap();
    b.group(dataWorkerGroup)
     .channel(NioSocketChannel.class)
     .option(ChannelOption.so_SNDBUF,262144)
     .option(ChannelOption.so_RCVBUF,262144)
     .option(ChannelOption.so_KEEPALIVE,true)
     .option(ChannelOption.TCP_NODELAY,true)
     .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,timeout)
     .handler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ch.config().setAllocator(UnpooledByteBufAllocator.DEFAULT);
            if(useSSL) {
                ch.pipeline().addLast(sslContext.newHandler(ch.alloc(),ip,dataPort));
            }
            ch.pipeline().addLast("encoder",new ByteArrayDecoder());
            ch.pipeline().addLast(new DataExceptionHandler());
            if(LOG.isTraceEnabled()) {
                ch.pipeline().addLast(new LoggingHandler(LogLevel.TRACE));
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) throws Exception {
            LOG.error("Cannot initialize data client.",getDataPort()).awaitUninterruptibly();

    if(future.isCancelled()) {
        dataChannel = null;
    } else if(!future.isSuccess()) {
        dataChannel = null;
        retryDataConnection();
    } else {
        dataChannel = future.channel();

        try {
            dataChannel.closeFuture().sync();
        } catch (InterruptedException ex) {
            LOG.debug("Interrupted waiting for client to shutdown.",ex);
        }
    }
}

private void initializeGossipClient() {
    LOG.trace("Initializing gossip client");

    gossipworkerGroup = new NioEventLoopGroup(GOSSIP_WORKER_THREADS);

    Bootstrap b = new Bootstrap();
    b.group(gossipworkerGroup);
    b.channel(NioSocketChannel.class);
    b.option(ChannelOption.so_KEEPALIVE,new ByteArrayDecoder());
            ch.pipeline().addLast(new GossipExceptionHandler());
            if (LOG.isTraceEnabled()) {
                ch.pipeline().addLast(new LoggingHandler(LogLevel.TRACE));
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,getGossipPort()).awaitUninterruptibly();

    if (future.isCancelled()) {
        gossipChannel = null;
    } else if (!future.isSuccess()) {
        gossipChannel = null;
        retryGossipConnection();
    } else {
        gossipChannel = future.channel();
        setStatus(MemberStatus.Alive);
        updateMember();
    }
}

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

    logger.debug("initChannel-start");

    ProtocolDecoderService protocolDecoderService = null;
    ProtocolEncoderService protocolEncoderService = null;

    try{
        protocolDecoderService = applicationContext.getBean(ProtocolDecoderService.class);
        protocolEncoderService = applicationContext.getBean(ProtocolEncoderService.class);

    }catch (Exception e){
        protocolDecoderService = new DefaultProtocolDecoderService();
        protocolEncoderService = new DefaultProtocolEncoderService();
    }

    logger.debug("initChannel->protocolDecoderService:"+protocolDecoderService);
    logger.debug("initChannel->protocolEncoderService:"+protocolEncoderService);


    ch.pipeline().addLast(ByteArrayDecoder,new ByteArrayDecoder());
    ch.pipeline().addLast(ByteArrayEncoder,new ByteArrayEncoder());

    ch.pipeline().addLast(LengthFieldBasedFrameDecoder,4));

    ch.pipeline().addLast(ProtocolDecoderHandler,new ProtocolDecoderHandler(protocolDecoderService));
    ch.pipeline().addLast(ProtocolEncoderHandler,new ProtocolEncoderHandler(protocolEncoderService));


    ch.pipeline().addLast(SystemTimeOut,new IdleStateHandler(heartTime,TimeUnit.SECONDS));

    ch.pipeline().addLast(SocketHandler,new SocketHandler(socketService));

    logger.debug("initChannel-end");
}

static void autoAddHttpExtractor(NettyContext c,String name,ChannelHandler
        handler){

    if (handler instanceof BytetoMessageDecoder
            || handler instanceof BytetoMessageCodec
            || handler instanceof CombinedChannelDuplexHandler) {
        String extractorName = name+"$extractor";

        if(c.channel().pipeline().context(extractorName) != null){
            return;
        }

        c.channel().pipeline().addBefore(name,extractorName,HTTP_EXTRACTOR);

        if(NettyContext.isPersistent(c.channel())){
            c.onClose(() -> c.removeHandler(extractorName));
        }

    }
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    UserConnection info = new UserConnection(socketChannel);
    // init protocol
    new ProtocolPipeline(info);
    // Add originals
    this.method.invoke(this.original,socketChannel);

    HandlerConstructor constructor = ClassGenerator.getConstructor();
    // Add our transformers
    MessagetoByteEncoder encoder = constructor.newEncodeHandler(info,(MessagetoByteEncoder) socketChannel.pipeline().get("encoder"));
    BytetoMessageDecoder decoder = constructor.newDecodeHandler(info,(BytetoMessageDecoder) socketChannel.pipeline().get("decoder"));
    BukkitPacketHandler chunkHandler = new BukkitPacketHandler(info);

    socketChannel.pipeline().replace("encoder","encoder",encoder);
    socketChannel.pipeline().replace("decoder","decoder",decoder);
    socketChannel.pipeline().addAfter("packet_handler","viaversion_packet_handler",chunkHandler);
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    // Ensure ViaVersion is loaded
    if (ProtocolRegistry.SERVER_PROTOCOL != -1) {
        UserConnection info = new UserConnection(socketChannel);
        // init protocol
        new ProtocolPipeline(info);
        // Add originals
        this.method.invoke(this.original,socketChannel);
        // Add our transformers
        MessagetoByteEncoder encoder = new SpongeEncodeHandler(info,(MessagetoByteEncoder) socketChannel.pipeline().get("encoder"));
        BytetoMessageDecoder decoder = new SpongeDecodeHandler(info,(BytetoMessageDecoder) socketChannel.pipeline().get("decoder"));
        SpongePacketHandler chunkHandler = new SpongePacketHandler(info);

        socketChannel.pipeline().replace("encoder",encoder);
        socketChannel.pipeline().replace("decoder",decoder);
        socketChannel.pipeline().addAfter("packet_handler",chunkHandler);
    } else {
        this.method.invoke(this.original,socketChannel);
    }
}

static void autoAddHttpExtractor(NettyContext c,String name,ChannelHandler
        handler){

    if (handler instanceof BytetoMessageDecoder
            || handler instanceof BytetoMessageCodec
            || handler instanceof CombinedChannelDuplexHandler) {
        String extractorName = name+"$extractor";

        if(c.channel().pipeline().context(extractorName) != null){
            return;
        }

        c.channel().pipeline().addBefore(name,extractorName,HTTP_EXTRACTOR);

        if(NettyContext.isPersistent(c.channel())){
            c.onClose(() -> c.removeHandler(extractorName));
        }

    }
}

ChannelHandler newSslInitiator() {
    return new BytetoMessageDecoder() {
        @Override
        protected void decode(ChannelHandlerContext ctx,ByteBuf in,List<Object> out) throws Exception {
            if(in.readableBytes() < 1) {
                return;
            }
            if('S' != in.readByte()) {
                ctx.fireExceptionCaught(new IllegalStateException("SSL required but not supported by backend server"));
                return;
            }
            ctx.pipeline().remove(this);
            ctx.pipeline().addFirst(
                    SslContextBuilder
                            .forClient()
                            .trustManager(InsecureTrustManagerFactory.INSTANCE)
                            .build()
                            .newHandler(ctx.alloc()));
        }
    };
}

/**
 * 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,8,-8,8);
}

@Override
protected void initChannel(Channel ch) throws Exception {
    ch.pipeline().addLast("openChannels",transport.serverOpenChannels);
    final HttpRequestDecoder decoder = new HttpRequestDecoder(
        Math.toIntExact(transport.maxInitialLineLength.getBytes()),Math.toIntExact(transport.maxHeaderSize.getBytes()),Math.toIntExact(transport.maxChunkSize.getBytes()));
    decoder.setCumulator(BytetoMessageDecoder.COMPOSITE_CUMULATOR);
    ch.pipeline().addLast("decoder",decoder);
    ch.pipeline().addLast("decoder_compress",new HttpContentDecompressor());
    ch.pipeline().addLast("encoder",new HttpResponseEncoder());
    final HttpObjectAggregator aggregator = new HttpObjectAggregator(Math.toIntExact(transport.maxContentLength.getBytes()));
    if (transport.maxCompositeBufferComponents != -1) {
        aggregator.setMaxCumulationBufferComponents(transport.maxCompositeBufferComponents);
    }
    ch.pipeline().addLast("aggregator",aggregator);
    if (transport.compression) {
        ch.pipeline().addLast("encoder_compress",new HttpContentCompressor(transport.compressionLevel));
    }
    if (SETTING_CORS_ENABLED.get(transport.settings())) {
        ch.pipeline().addLast("cors",new Netty4CorsHandler(transport.getCorsConfig()));
    }
    if (transport.pipelining) {
        ch.pipeline().addLast("pipelining",new HttpPipeliningHandler(transport.pipeliningMaxEvents));
    }
    ch.pipeline().addLast("handler",requestHandler);
}

/**
 * Call the decode method on a netty BytetoMessageDecoder
 *
 * @param decoder The decoder
 * @param ctx     The current context
 * @param input   The packet to decode
 * @return A list of the decoders output
 * @throws InvocationTargetException If an exception happens while executing
 */
public static List<Object> callDecode(BytetoMessageDecoder decoder,ChannelHandlerContext ctx,Object input) throws InvocationTargetException {
    List<Object> output = new ArrayList<>();
    try {
        PipelineUtil.DECODE_METHOD.invoke(decoder,ctx,input,output);
    } catch (illegalaccessexception e) {
        e.printstacktrace();
    }
    return output;
}

@Override
public BytetoMessageDecoder getDecoder() {
    ByteDecoder decoder = new ByteDecoder();
    decoder.setMessageCodec(msgCodec);
    decoder.setHeadCodec(headCodec);
    return decoder;
}

protected void initChannel(final Channel channel) throws Exception
{
    super.initChannel(channel);
    channel.pipeline().addFirst("sslDetectionHandler",new BytetoMessageDecoder()
    {
        @Override
        protected void decode(ChannelHandlerContext channelHandlerContext,ByteBuf byteBuf,List<Object> list) throws Exception
        {
            if (byteBuf.readableBytes() < 5)
            {
                // To detect if SSL must be used we need to have at least 5 bytes,so return here and try again
                // once more bytes a ready.
                return;
            }
            if (SslHandler.isEncrypted(byteBuf))
            {
                // Connection uses SSL/TLS,replace the detection handler with a SslHandler and so use
                // encryption.
                SslHandler sslHandler = createSslHandler();
                channelHandlerContext.pipeline().replace(this,"ssl",sslHandler);
            }
            else
            {
                // Connection use no TLS/SSL encryption,just remove the detection handler and continue without
                // SslHandler in the pipeline.
                channelHandlerContext.pipeline().remove(this);
            }
        }
    });
}

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

    // Initialize our session Object when the channel is initialized,attach
    // it to the channel.
    ch.attr(NetworkConstants.SESSION_KEY).setIfAbsent(new PlayerIO(ch));

    // Initialize the pipeline channel handlers.
    ChannelDuplexHandler timeout = new IdleStateHandler(NetworkConstants.INPUT_TIMEOUT,0);
    BytetoMessageDecoder loginHandshakeHandler = new LoginHandshakeHandler();

    ch.pipeline().addLast("login-handshake",loginHandshakeHandler);
    ch.pipeline().addLast("channel-handler",channelHandler);
    ch.pipeline().addLast("timeout",timeout);
}

@Override
public BukkitDecodeHandler newDecodeHandler(UserConnection info,BytetoMessageDecoder minecraftDecoder) {
    return new BukkitDecodeHandler(info,minecraftDecoder);
}

public BukkitDecodeHandler(UserConnection info,BytetoMessageDecoder minecraftDecoder) {
    this.info = info;
    this.minecraftDecoder = minecraftDecoder;
}

public SpongeDecodeHandler(UserConnection info,BytetoMessageDecoder minecraftDecoder) {
    this.info = info;
    this.minecraftDecoder = minecraftDecoder;
}

/**
 * 获取解码器 @see BytetoMessageDecoder
 * 
 * @return 解码器
 */
BytetoMessageDecoder getDecoder();

public BytetoMessageDecoder newDecodeHandler(UserConnection info,BytetoMessageDecoder minecraftDecoder);

private static void addCompression(ChannelHandlerContext ctx) {
    ctx.pipeline().addFirst(new JdkZlibEncoder(),new JdkZlibDecoder());
}