java类io.netty.handler.codec.http.HttpObjectAggregator的实例源码

@Override
protected void initChannel(Channel ch) throws Exception {
    // create a new pipeline
    ChannelPipeline pipeline = ch.pipeline();

    SslHandler sslHandler = configureServerSSLOnDemand();
    if (sslHandler != null) {
        LOG.debug("Server SSL handler configured and added as an interceptor against the ChannelPipeline: {}", sslHandler);
        pipeline.addLast("ssl", sslHandler);
    }

    pipeline.addLast("decoder", new HttpRequestDecoder(409, configuration.getMaxHeaderSize(), 8192));
    pipeline.addLast("encoder", new HttpResponseEncoder());
    if (configuration.isChunked()) {
        pipeline.addLast("aggregator", new HttpObjectAggregator(configuration.getChunkedMaxContentLength()));
    }
    if (configuration.isCompression()) {
        pipeline.addLast("deflater", new HttpContentCompressor());
    }

    pipeline.addLast("handler", channelFactory.getChannelHandler());
}

public final ChannelPipeline appendHttpPipeline(ChannelPipeline channelPipeline) {
    // 服务端，对响应编码。属于ChannelOutboundHandler，逆序执行
    channelPipeline.addLast("encoder", new HttpResponseEncoder());

    // 服务端，对请求解码。属于ChannelIntboundHandler，按照顺序执行
    channelPipeline.addLast("decoder", new HttpRequestDecoder());
    //即通过它可以把 HttpMessage 和 HttpContent 聚合成一个 FullHttpRequest,并定义可以接受的数据大小，在文件上传时，可以支持params+multipart
    channelPipeline.addLast("aggregator", new HttpObjectAggregator(maxConentLength));
    //块写入写出Handler
    channelPipeline.addLast("chunkedWriter", new ChunkedWriteHandler());

    // 对传输数据进行压缩，这里在客户端需要解压缩处理
    // channelPipeline.addLast("deflater", new HttpContentCompressor());

    HttpServletHandler servletHandler = new HttpServletHandler();
    servletHandler.addInterceptor(new ChannelInterceptor());
    //servletHandler.addInterceptor(new HttpSessionInterceptor(getHttpSessionStore()));
    // 自定义Handler
    channelPipeline.addLast("handler", servletHandler);
    // 异步
    // channelPipeline.addLast(businessExecutor, new AsyncHttpServletHandler());
    return channelPipeline;
}

@Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline p = ch.pipeline();
        if(sslCtx!=null)
        {
            p.addLast(new SslHandler(sslCtx.newEngine(ch.alloc())));
        }
        p.addLast(new HttpResponseEncoder());//必须放在最前面,如果decoder途中需要回复消息,则decoder前面需要encoder
        p.addLast(new HttpRequestDecoder());
        p.addLast(new HttpObjectAggregator(65536));//限制contentLength
        //大文件传输处理
//      p.addLast(new ChunkedWriteHandler());
//      p.addLast(new HttpContentCompressor());
        //跨域配置
        CorsConfig corsConfig = CorsConfigBuilder.forAnyOrigin().allowNullOrigin().allowCredentials().build();
        p.addLast(new CorsHandler(corsConfig));
        p.addLast(new DefaultListenerHandler<HttpRequest>(listener));
    }

/**
     * 适配
     */
    @Override
    protected ChannelHandler fixHandlerBeforeConnect(final ChannelHandler handler) {
        ChannelHandler result=new ShareableChannelInboundHandler() {
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                Channel ch=ctx.channel();
                ch.pipeline().addLast(new HttpClientCodec());
                ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
                ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
                ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
                ctx.pipeline().remove(this);//移除当前handler
                ctx.pipeline().fireChannelRegistered();//重新从第一个handler抛出事件
            }
        };
//      ChannelInitializer<SocketChannel> result=new ChannelInitializer<SocketChannel>() {
//            @Override
//            protected void initChannel(SocketChannel ch) {
//              ch.pipeline().addLast(new HttpClientCodec());
//              ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
//              ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
//              ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
//            }
//        };
        return result;
    }

/**
     * 适配
     */
    @Override
    protected ChannelHandler fixHandlerBeforeConnect(final ChannelHandler handler) {
        ChannelHandler result=new ShareableChannelInboundHandler() {
            @Override
            public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
                Channel ch=ctx.channel();
                ch.pipeline().addLast(new HttpClientCodec());
                ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
                ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
                ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
                ctx.pipeline().remove(this);//移除当前handler
                ctx.fireChannelRegistered();//重新从第一个handler抛出事件
            }
        };
//      ChannelInitializer<SocketChannel> result=new ChannelInitializer<SocketChannel>() {
//            @Override
//            protected void initChannel(SocketChannel ch) {
//              ch.pipeline().addLast(new HttpClientCodec());
//              ch.pipeline().addLast(new HttpObjectAggregator(64*1024));
//              ch.pipeline().addLast(new WebSocketClientProtocolHandler(WebSocketClientHandshakerFactory.newHandshaker(uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders())));
//              ch.pipeline().addLast(new WebSocketConnectedClientHandler(handler));
//            }
//        };
        return result;
    }

@Override
public void init() {
    super.init();
    b.group(bossGroup, workGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.SO_KEEPALIVE, false)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_BACKLOG, 1024)
            .localAddress(new InetSocketAddress(port))
            .childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(defLoopGroup,
                            new HttpRequestDecoder(),       //请求解码器
                            new HttpObjectAggregator(65536),//将多个消息转换成单一的消息对象
                            new HttpResponseEncoder(),      // 响应编码器
                            new HttpServerHandler(snowFlake)//自定义处理器
                    );
                }
            });

}

protected ChannelHandler setupWSChannel(SslContext sslCtx, Configuration conf, DataStore datastore) {
    return new ChannelInitializer<SocketChannel>() {

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast("ssl", sslCtx.newHandler(ch.alloc()));
            ch.pipeline().addLast("httpServer", new HttpServerCodec());
            ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
            ch.pipeline().addLast("sessionExtractor", new WebSocketHttpCookieHandler(config));
            ch.pipeline().addLast("idle-handler", new IdleStateHandler(conf.getWebsocket().getTimeout(), 0, 0));
            ch.pipeline().addLast("ws-protocol", new WebSocketServerProtocolHandler(WS_PATH, null, true));
            ch.pipeline().addLast("wsDecoder", new WebSocketRequestDecoder(datastore, config));
            ch.pipeline().addLast("error", new WSExceptionHandler());
        }
    };

}

@Override
        protected void initChannel(SocketChannel ch) throws Exception {
            try {
                ch.config().setOption(ChannelOption.IP_TOS, 0x18);
//                ch.config().setOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 32 * 1024);
//                ch.config().setOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, 8 * 1024);
            } catch (ChannelException ex) {
                // IP_TOS not supported by platform, ignore
            }
            ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);

            PacketRegistry r = new PacketRegistry();

            ch.pipeline().addLast(new HttpServerCodec());
            ch.pipeline().addLast(new HttpObjectAggregator(65536));
            ch.pipeline().addLast(new WebSocketHandler());
            ch.pipeline().addLast(new PacketDecoder(r));
            ch.pipeline().addLast(new PacketEncoder(r));
            ch.pipeline().addLast(new ClientHandler(server));
        }

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    try {
        ch.config().setOption(ChannelOption.TCP_NODELAY, true);
        ch.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException ex) {
        // IP_TOS not supported by platform, ignore
    }
    ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    PacketRegistry r = new PacketRegistry();

    ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(0, 0, 30));
    ch.pipeline().addLast(new HttpServerCodec());
    ch.pipeline().addLast(new HttpObjectAggregator(65536));
    ch.pipeline().addLast(new WebSocketHandler());
    ch.pipeline().addLast(new PacketDecoder(r));
    ch.pipeline().addLast(new PacketEncoder(r));
    ch.pipeline().addLast(mExecutorGroup, "serverHandler", new ClientHandler(mServer));
}

@Override
protected void doStart(Listener listener) throws Throwable {
    workerGroup = new NioEventLoopGroup(http_work, new DefaultThreadFactory(ThreadNames.T_HTTP_CLIENT));
    b = new Bootstrap();
    b.group(workerGroup);
    b.channel(NioSocketChannel.class);
    b.option(ChannelOption.SO_KEEPALIVE, true);
    b.option(ChannelOption.TCP_NODELAY, true);
    b.option(ChannelOption.SO_REUSEADDR, true);
    b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 4000);
    b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    b.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast("decoder", new HttpResponseDecoder());
            ch.pipeline().addLast("aggregator", new HttpObjectAggregator(maxContentLength));
            ch.pipeline().addLast("encoder", new HttpRequestEncoder());
            ch.pipeline().addLast("handler", new HttpClientHandler(NettyHttpClient.this));
        }
    });
    timer = new HashedWheelTimer(new NamedThreadFactory(T_HTTP_TIMER), 1, TimeUnit.SECONDS, 64);
    listener.onSuccess();
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    if (msg instanceof ByteBuf && ctx.channel().isActive()) {
        boolean isHttpRequest = false;
        ByteBuf buffer = (ByteBuf) msg;
        final int len = 11;
        if (buffer.readableBytes() > len) {
            byte[] dst = new byte[len];
            buffer.getBytes(buffer.readerIndex(), dst, 0, len);
            int n = HttpMethodUtil.method(dst);
            isHttpRequest = n > 2;
        }
        if (isHttpRequest) {
            ChannelPipeline cp = ctx.pipeline();
            String currentName = ctx.name();
            cp.addAfter(currentName, "HttpRequestDecoder", new HttpRequestDecoder());
            cp.addAfter("HttpRequestDecoder", "HttpResponseEncoder", new HttpResponseEncoder());
            cp.addAfter("HttpResponseEncoder", "HttpObjectAggregator", new HttpObjectAggregator(512 * 1024));
            ChannelHandler handler = serverDef.httpHandlerFactory.create(serverDef);
            cp.addAfter("HttpObjectAggregator", "HttpThriftBufDecoder", handler);

            cp.remove(currentName);
        }
    }
    ctx.fireChannelRead(msg);
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    try {
        ch.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException ex) {
        // IP_TOS not supported by platform, ignore
    }
    ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    ch.pipeline().addLast(new HttpServerCodec());
    ch.pipeline().addLast(new HttpObjectAggregator(65536));
    ch.pipeline().addLast(new WebSocketHandler());
    ch.pipeline().addLast(new PacketDecoder());
    ch.pipeline().addLast(new PacketEncoder());
    ch.pipeline().addLast(new ClientHandler(server));
}

@Override
protected void initChannel(SocketChannel channel) throws SSLException {
    URI uri = config.getConnectionWebsocketUri();

    DefaultHttpHeaders headers = new DefaultHttpHeaders();
    headers.add(USER_ID_HEADER, config.getConnectionUserId().toString());
    headers.add(USER_PASSWORD_HEADER, config.getConnectionUserPassword());
    headers.add(SUPPLIER_ID_HEADER, config.getConnectionServerId());

    WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(uri, WS_VERSION, null, false, headers);

    ChannelPipeline pipeline = channel.pipeline();
    if (config.isConnectionSecure()) {
        try {
            SslContext sslContext = SslContext.newClientContext(InsecureTrustManagerFactory.INSTANCE);
            pipeline.addLast(sslContext.newHandler(channel.alloc()));
        } catch (SSLException e) {
            logger.log(Level.SEVERE, "Shutting down client due to unexpected failure to create SSL context", e);
            throw e;
        }
    }
    pipeline.addLast(new HttpClientCodec());
    pipeline.addLast(new HttpObjectAggregator(8192));
    pipeline.addLast(new AudioConnectClientHandler(handshaker));
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    try {
        ch.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException ex) {
        // IP_TOS not supported by platform, ignore
    }
    ch.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    ch.pipeline().addLast(new HttpServerCodec());
    ch.pipeline().addLast(new HttpObjectAggregator(65536));
    ch.pipeline().addLast(new Handshaker());
    ch.pipeline().addLast(new WebSocketHandler());
    ch.pipeline().addLast(new PacketDecoder());
    ch.pipeline().addLast(new PacketEncoder());
    ch.pipeline().addLast(new ClientHandler(server));
}

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    //编解码http请求
    pipeline.addLast(new HttpServerCodec());
    //聚合解码HttpRequest/HttpContent/LastHttpContent到FullHttpRequest
    //保证接收的Http请求的完整性
    pipeline.addLast(new HttpObjectAggregator(64 *1024));
    //写文件内容
    pipeline.addLast(new ChunkedWriteHandler());
    //处理FullHttpRequest
    pipeline.addLast(new HttpRequestHandler("/ws"));
    //处理其他的WebSocketFrame
    pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    //处理TextWebSocketFrame
    pipeline.addLast(new TextWebSocketFrameHandler(group));
}

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    //编解码http请求
    pipeline.addLast(new HttpServerCodec());
    //聚合解码HttpRequest/HttpContent/LastHttpContent到FullHttpRequest
    //保证接收的Http请求的完整性
    pipeline.addLast(new HttpObjectAggregator(64 *1024));
    //写文件内容
    pipeline.addLast(new ChunkedWriteHandler());
    //处理FullHttpRequest
    pipeline.addLast(new HttpRequestHandler("/ws"));
    //处理其他的WebSocketFrame
    pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    //处理TextWebSocketFrame
    pipeline.addLast(new TextWebSocketFrameHandler(group));
}

public static Bootstrap createNettyHttpClientBootstrap() {
    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(new NioEventLoopGroup())
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new HttpClientCodec());
                     p.addLast(new HttpObjectAggregator(Integer.MAX_VALUE));
                     p.addLast("clientResponseHandler", new SimpleChannelInboundHandler<HttpObject>() {
                         @Override
                         protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
                             throw new RuntimeException("Client response handler was not setup before the call");
                         }
                     });
                 }
             });

    return bootstrap;
}

@Override
public void configure(final ChannelPipeline pipeline) {
    final String scheme = connection.getUri().getScheme();
    if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("Unsupported scheme (only ws: or wss: supported): " + scheme);

    if (!supportsSsl() && "wss".equalsIgnoreCase(scheme))
        throw new IllegalStateException("To use wss scheme ensure that enableSsl is set to true in configuration");

    final int maxContentLength = cluster.connectionPoolSettings().maxContentLength;
    handler = new WebSocketClientHandler(
            WebSocketClientHandshakerFactory.newHandshaker(
                    connection.getUri(), WebSocketVersion.V13, null, false, HttpHeaders.EMPTY_HEADERS, maxContentLength));

    pipeline.addLast("http-codec", new HttpClientCodec());
    pipeline.addLast("aggregator", new HttpObjectAggregator(maxContentLength));
    pipeline.addLast("ws-handler", handler);
    pipeline.addLast("gremlin-encoder", webSocketGremlinRequestEncoder);
    pipeline.addLast("gremlin-decoder", webSocketGremlinResponseDecoder);
}

@Override
public void configure(final ChannelPipeline pipeline) {
    if (logger.isDebugEnabled())
        pipeline.addLast(new LoggingHandler("log-io", LogLevel.DEBUG));

    pipeline.addLast("http-server", new HttpServerCodec());

    if (logger.isDebugEnabled())
        pipeline.addLast(new LoggingHandler("http-io", LogLevel.DEBUG));

    pipeline.addLast(new HttpObjectAggregator(settings.maxContentLength));

    if (authenticator != null) {
        // Cannot add the same handler instance to multiple times unless
        // it is marked as @Sharable, indicating a race condition will
        // not occur. It may not be a safe assumption that the handler
        // is sharable so create a new handler each time.
        authenticationHandler = authenticator.getClass() == AllowAllAuthenticator.class ?
                null : new HttpBasicAuthenticationHandler(authenticator);
        if (authenticationHandler != null)
            pipeline.addLast(PIPELINE_AUTHENTICATOR, authenticationHandler);
    }

    pipeline.addLast("http-gremlin-handler", httpGremlinEndpointHandler);
}

@Override
public void onOpen(Session session, EndpointConfig config) {
    String id = session.getId();
    log.debug("{}: open ws proxy ", id);
    try {
        ChannelFuture cf = backend.connect().sync();
        Channel channel = cf.channel();
        WebSocketClientProtocolHandler wscph = makeWsProtocolHandler(session);
        WebSocketClientHandshaker handshaker = wscph.handshaker();
        WsHandler handler = new WsHandler(handshaker, channel, session);
        channel.pipeline().addLast(new HttpObjectAggregator(1024 * 4),
          WebSocketClientCompressionHandler.INSTANCE,
          wscph,
          handler);
        handshaker.handshake(channel);
        log.debug("{}: wait messages", id);
        session.addMessageHandler(String.class, handler::onFrontString);
        session.addMessageHandler(ByteBuffer.class, handler::onFrontBytes);
    } catch (Exception e) {
        log.error("{}: can not establish ws connect with backed", id, e);
    }

}

@Override
protected void configurePipeline(ChannelHandlerContext ctx, String protocol) throws Exception {
    if (ApplicationProtocolNames.HTTP_2.equals(protocol)) {
        ctx.pipeline().addLast(new Http2MultiplexCodec(true, new HelloWorldHttp2Handler()));
        return;
    }

    if (ApplicationProtocolNames.HTTP_1_1.equals(protocol)) {
        ctx.pipeline().addLast(new HttpServerCodec(),
                               new HttpObjectAggregator(MAX_CONTENT_LENGTH),
                               new HelloWorldHttp1Handler("ALPN Negotiation"));
        return;
    }

    throw new IllegalStateException("unknown protocol: " + protocol);
}

@Override
protected void configurePipeline(ChannelHandlerContext ctx, String protocol) throws Exception {
    if (ApplicationProtocolNames.HTTP_2.equals(protocol)) {
        ctx.pipeline().addLast(new HelloWorldHttp2HandlerBuilder().build());
        return;
    }

    if (ApplicationProtocolNames.HTTP_1_1.equals(protocol)) {
        ctx.pipeline().addLast(new HttpServerCodec(),
                               new HttpObjectAggregator(MAX_CONTENT_LENGTH),
                               new HelloWorldHttp1Handler("ALPN Negotiation"));
        return;
    }

    throw new IllegalStateException("unknown protocol: " + protocol);
}

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

        // Enable HTTPS if necessary.
//      if (sslCtx != null) {
//          pipeline.addLast(sslCtx.newHandler(ch.alloc()));
//      }

        pipeline.addLast(new HttpClientCodec());
        // Remove the following line if you don't want automatic content decompression.
        pipeline.addLast(new HttpContentDecompressor());

        // Uncomment the following line if you don't want to handle HttpContents.
        //pipeline.addLast(new HttpObjectAggregator(65536));
        pipeline.addLast(new HttpObjectAggregator(65536 * 3));


//      pipeline.addLast(new HttpClientHandler(null, mHttpClientListener));
    }

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    if (sslCtx != null) {
        pipeline.addLast(sslCtx.newHandler(ch.alloc()));
    }
    pipeline.addLast(new HttpResponseEncoder());
    pipeline.addLast(new HttpRequestDecoder());
    // Uncomment the following line if you don't want to handle HttpChunks.
    //pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());
    //p.addLast(new HttpObjectAggregator(1048576));
    // Remove the following line if you don't want automatic content compression.
    //pipeline.addLast(new HttpContentCompressor());

    // Uncomment the following line if you don't want to handle HttpContents.
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast("readTimeoutHandler", new ReadTimeoutHandler(READ_TIMEOUT));
    pipeline.addLast("myHandler", new MyHandler());

    pipeline.addLast("handler", new HttpServerHandler(listener));
}

public void start(int port) throws Exception {
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast("http-decoder", new HttpRequestDecoder());
                ch.pipeline().addLast("http-aggregator", new HttpObjectAggregator(65536));
                ch.pipeline().addLast("http-encoder", new HttpResponseEncoder());
                ch.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
                ch.pipeline().addLast("serverHandler", new HttpServerHandler());
            }
        }).option(ChannelOption.SO_BACKLOG, 1024).childOption(ChannelOption.SO_KEEPALIVE, true);

        ChannelFuture f = b.bind(port).sync();

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

@Provides
@HttpsRelayProtocol
static ImmutableList<Provider<? extends ChannelHandler>> provideHandlerProviders(
    Provider<SslClientInitializer<NioSocketChannel>> sslClientInitializerProvider,
    Provider<HttpClientCodec> httpClientCodecProvider,
    Provider<HttpObjectAggregator> httpObjectAggregatorProvider,
    Provider<BackendMetricsHandler> backendMetricsHandlerProvider,
    Provider<LoggingHandler> loggingHandlerProvider,
    Provider<FullHttpResponseRelayHandler> relayHandlerProvider) {
  return ImmutableList.of(
      sslClientInitializerProvider,
      httpClientCodecProvider,
      httpObjectAggregatorProvider,
      backendMetricsHandlerProvider,
      loggingHandlerProvider,
      relayHandlerProvider);
}

public static void addHttpServerCodec(@NotNull ChannelPipeline pipeline) {
  pipeline.addLast("httpRequestEncoder", new HttpResponseEncoder());
  // https://jetbrains.zendesk.com/agent/tickets/68315
  pipeline.addLast("httpRequestDecoder", new HttpRequestDecoder(16 * 1024, 16 * 1024, 8192));
  pipeline.addLast("httpObjectAggregator", new HttpObjectAggregator(MAX_CONTENT_LENGTH));
  // could be added earlier if HTTPS
  if (pipeline.get(ChunkedWriteHandler.class) == null) {
    pipeline.addLast("chunkedWriteHandler", new ChunkedWriteHandler());
  }
  pipeline.addLast("corsHandler", new CorsHandlerDoNotUseOwnLogger(CorsConfig
                                                                     .withAnyOrigin()
                                                                     .allowCredentials()
                                                                     .allowNullOrigin()
                                                                     .allowedRequestMethods(HttpMethod.GET, HttpMethod.POST, HttpMethod.PUT, HttpMethod.DELETE, HttpMethod.HEAD, HttpMethod.PATCH)
                                                                     .allowedRequestHeaders("origin", "accept", "authorization", "content-type")
                                                                     .build()));
}

public static void main(String[] args) {
    String ip = "127.0.0.1";
    int port = 8080;
    ChannelInitializer<SocketChannel> channelInit = new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();

            CorsConfig corsConfig = CorsConfig.withAnyOrigin()
                    .allowedRequestHeaders("content-type","accept","MyCustomHeader")
                    .allowedRequestMethods(PUT,POST,GET,DELETE)
                    .build();

            p.addLast(new HttpResponseEncoder());
            p.addLast(new HttpRequestDecoder());
            p.addLast(new HttpObjectAggregator(65536));
            p.addLast(new ChunkedWriteHandler());
            p.addLast(new CorsHandler(corsConfig));
            p.addLast(new SimpleCORSHandler());
        }
    };
    NettyServerUtil.newHttpServerBootstrap(ip, port, channelInit);
}

@Override
protected void decode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out)
{
    if(!added && msg instanceof HttpRequest)
    {
        String path = ((HttpRequest)msg).getUri();
        WsServerHandler handler = findHandler(path);
        if(handler != null)
        {
            ctx.pipeline().addAfter("switch", "aggregator", new HttpObjectAggregator(65536));
            ctx.pipeline().addAfter("aggregator", "wsprotocol", new WebSocketServerProtocolHandler(path, null, true));
            ctx.pipeline().addAfter("wsprotocol", "wshandler", new WsFrameHandler(handler));
            added = true;
        }
    }
    ReferenceCountUtil.retain(msg);
    out.add(msg);
}

@Override
public void configNewChannel(NioSocketChannel channel) {
    super.configNewChannel(channel);
    ChannelPipeline pipeline = channel.pipeline();
    // 添加 SSL 数据支持
    if (requestConfig.https()) {
        SslContext sslContent = NettyCenter.singleInstance().getSimpleClientSslContext();
        SSLEngine engine = sslContent.newEngine(channel.alloc());
        pipeline.addLast("ssl", new SslHandler(engine));
    }
    // 客户端接收到的是httpResponse响应，所以要使用HttpResponseDecoder进行解码
    pipeline.addLast("decoder", new HttpResponseDecoder());
    // 客户端发送的是httprequest，所以要使用HttpRequestEncoder进行编码
    pipeline.addLast("encoder", new HttpRequestEncoder());
    // 接收的请求累计器
    pipeline.addLast("aggegator", new HttpObjectAggregator(0x30000));
    // mime 类型写出
    pipeline.addLast("streamew", new ChunkedWriteHandler());
    // 添加解压器
    pipeline.addLast("decompressor", new HttpContentDecompressor());
    // add new handler
    pipeline.addLast("handler", new NettyHttpRequestChannelHandler());
}