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

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

    pipeline.addLast(new HttpRequestDecoder());
    pipeline.addLast(new HttpResponseEncoder());

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

    pipeline.addLast(new HttpUploadServerHandler());
}
 

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    //添加闲置处理,60秒没有数据传输，触发事件
    pipeline.addLast(new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS));
    //将字节解码为HttpMessage对象，并将HttpMessage对象编码为字节
    pipeline.addLast(new HttpServerCodec());
    //出站数据压缩
    pipeline.addLast(new HttpContentCompressor());
    //聚合多个HttpMessage为单个FullHttpRequest
    pipeline.addLast(new HttpObjectAggregator(64 * 1024));
    //如果被请求的端点是/ws，则处理该升级握手
    pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    //聊天消息处理
    pipeline.addLast(new ChatServerHandler());
    //心跳处理
    pipeline.addLast(new HeartbeatHandler());
}
 

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

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

    pipeline.addLast(new HttpRequestDecoder());
    pipeline.addLast(new HttpResponseEncoder());

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

    pipeline.addLast(new HttpUploadServerHandler());
}
 

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

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

    pipeline.addLast(new HttpRequestDecoder());
    pipeline.addLast(new HttpResponseEncoder());

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

    pipeline.addLast(new HttpUploadServerHandler());
}
 

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }

    p.addLast(new HttpRequestDecoder());
    // Uncomment the following line if you don't want to handle HttpChunks.
    p.addLast(new HttpObjectAggregator(65536));
    p.addLast(new HttpResponseEncoder());
    // Remove the following line if you don't want automatic content
    // compression.
    p.addLast(new HttpContentCompressor());
    p.addLast(new ApiRequestParser());
}
 

@Override
protected void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    try {
        if (sslCtx != null) {
            pipeline.addLast(sslCtx.newHandler(ch.alloc()));
        }

        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new HttpServerExpectContinueHandler());

        if (useGZIP) {
            pipeline.addLast(new HttpContentCompressor());
        }

        if (isWebSocket) {
            pipeline.addLast(new WebSocketHandler(blade));
        }
        pipeline.addLast(new MergeRequestHandler());
        pipeline.addLast(httpServerHandler);
    } catch (Exception e) {
        log.error("Add channel pipeline error", e);
    }
}
 

@Override
 protected void initChannel(Channel channel) throws Exception {
// Create a default pipeline implementation.
   ChannelPipeline pipeline = channel.pipeline();

   // Uncomment the following line if you want HTTPS
   // SSLEngine engine =
   // SecureChatSslContextFactory.getServerContext().createSSLEngine();
   // engine.setUseClientMode(false);
   // pipeline.addLast("ssl", new SslHandler(engine));

   pipeline.addLast("decoder", new HttpRequestDecoder());
   //pipeline.addLast("aggregator", new HttpChunkAggregator(65536));
   pipeline.addLast("encoder", new HttpResponseEncoder());
   pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());
   pipeline.addLast("deflater", new HttpContentCompressor());
   pipeline.addLast("handler", new HttpDataServerHandler(userName, appId));
 }
 

@Override
public void initChannel(SocketChannel ch) {
	ChannelPipeline p = ch.pipeline();
	// p.addLast("log", new LoggingHandler(LogLevel.ERROR));

	if (sslCtx != null) {
		p.addLast(sslCtx.newHandler(ch.alloc()));
	}
	p.addLast(new HttpRequestDecoder());
	p.addLast(new HttpResponseEncoder());
	p.addLast("http compressor", new HttpContentCompressor());
	// p.addLast(new HttpObjectAggregator(1048576));
	p.addLast("http dispatcher", reqDis);
	p.addLast("idleStateHandler", new IdleStateHandler(10, 10, 0));
	p.addLast("heartbeatHandler", new HeartbeatHandler());
}
 

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();
    // p.addLast("log", new LoggingHandler(LogLevel.ERROR));

    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    p.addLast(new HttpRequestDecoder());
    p.addLast(new HttpResponseEncoder());
    p.addLast("http compressor", new HttpContentCompressor());
    // p.addLast(new HttpObjectAggregator(1048576));
    p.addLast("http dispatcher", reqDis);
    p.addLast("idleStateHandler", new IdleStateHandler(30, 10, 0));
    p.addLast("heartbeatHandler", new HeartbeatHandler());
}
 

protected ChannelPipeline getDefaulHttpChannelPipeline(Channel channel) throws Exception {

        // Create a default pipeline implementation.
        ChannelPipeline pipeline = channel.pipeline();

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

        pipeline.addLast("decoder", new HttpRequestDecoder());
        pipeline.addLast("encoder", new HttpResponseEncoder());
        pipeline.addLast("aggregator", new HttpObjectAggregator(maxChunkContentSize));
        
        // Remove the following line if you don't want automatic content
        // compression.
        pipeline.addLast("deflater", new HttpContentCompressor());
        // Set up the idle handler
        pipeline.addLast("idle", new IdleStateHandler(nettyHttpServerEngine.getReadIdleTime(),
                nettyHttpServerEngine.getWriteIdleTime(), 0));

        return pipeline;
    }
 

public void run() throws InterruptedException {
    // Configure the server.
    ServerBootstrap bootstrap = new ServerBootstrap();
    group = new NioEventLoopGroup();
    bootstrap.group(group);
    bootstrap.channel(NioServerSocketChannel.class);
    bootstrap.childHandler(new ChannelInitializer<Channel>() {

        @Override
        protected void initChannel(Channel ch) {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast("decoder", new HttpRequestDecoder());
            pipeline.addLast("encoder", new HttpResponseEncoder());
            pipeline.addLast("deflater", new HttpContentCompressor());
            pipeline.addLast("handler", new HttpTestServerHandler());
        }
    });

    // Bind and start to accept incoming connections.
    channel = bootstrap.bind(new InetSocketAddress(port)).sync().channel();
}
 

@Override
protected void initSocketChannel(SocketChannel ch) {
	super.initSocketChannel(ch);
	ch.pipeline().addBefore(HANDLER_MQTTDECODER, "HttpServerCodec", new HttpServerCodec());
	ch.pipeline().addBefore(HANDLER_MQTTDECODER, "HttpObjectAggregator", new HttpObjectAggregator(65536));
	ch.pipeline().addBefore(HANDLER_MQTTDECODER, "ChunkedWriteHandler", new ChunkedWriteHandler());
	ch.pipeline().addBefore(HANDLER_MQTTDECODER, "compressor ", new HttpContentCompressor());
	ch.pipeline().addBefore(HANDLER_MQTTDECODER, "protocol", new WebSocketServerProtocolHandler(websocketPath, "mqtt,mqttv3.1,mqttv3.1.1", true, 65536));
	ch.pipeline().addBefore(HANDLER_MQTTDECODER, "mqttWebSocket", new MqttWebSocketCodec());

	HttpResourceHander httpResourceHander = new HttpResourceHander(httpResource);
	ch.pipeline().addLast("httpResource", httpResourceHander);
}
 

private void switchToHttp(ChannelHandlerContext ctx) {
    ChannelPipeline p = ctx.pipeline();
    p.addLast("decoder", new HttpRequestDecoder());
    p.addLast("encoder", new HttpResponseEncoder());
    p.addLast("deflater", new HttpContentCompressor());
    p.addLast("handler", new HttpSnoopServerHandler());
    p.remove(this);
}
 

@Override
public void initChannel(final SocketChannel ch) {
    ch.pipeline()
            .addLast(new HttpServerCodec())
            .addLast(new HttpObjectAggregator(1048576))
            .addLast(new HttpContentCompressor())
            .addLast(new ChunkedWriteHandler())
            .addLast(new HttpHandler(harvester, helpExposition));
}
 

@Override
protected void initChannel(SocketChannel socketChannel) {
    //channel 代表了一个socket.
    ChannelPipeline pipeline = socketChannel.pipeline();
    pipeline.addLast(new ReadTimeoutHandler(1));

    /**
     * http-request解码器
     * http服务器端对request解码
     */
    pipeline.addLast("decoder", new HttpRequestDecoder(8192, 8192, 8192));
    /**
     * http-response解码器
     * http服务器端对response编码
     */
    pipeline.addLast("encoder", new HttpResponseEncoder());

    /**
     * 把多个消息转换为一个单一的FullHttpRequest或是FullHttpResponse
     */
    pipeline.addLast("aggregator", new HttpObjectAggregator(10 * 1024 * 1024));

    /**
     * 压缩
     */
    pipeline.addLast(new HttpContentCompressor());

    /**
     * handler分为两种，inbound handler,outbound handler,分别处理 流入，流出。
     * 服务端业务逻辑
     */
    pipeline.addLast(new HttpServerHandler());

}
 

/**
 * 可以在 HttpServerCodec 之后添加这些 ChannelHandler 进行开启高级特性
 */
private void addAdvanced(ChannelPipeline pipeline){
    if(CommonConstants.USE_COMPRESS) {
        // 对 http 响应结果开启 gizp 压缩
        pipeline.addLast(new HttpContentCompressor());
    }
    if(CommonConstants.USE_AGGREGATOR) {
        // 将多个HttpRequest组合成一个FullHttpRequest
        pipeline.addLast(new HttpObjectAggregator(CommonConstants.MAX_CONTENT_LENGTH));
    }
}
 

/**
 * 可以在 HttpServerCodec 之后添加这些 ChannelHandler 进行开启高级特性
 */
private void addAdvanced(ChannelPipeline pipeline){
    if(CommonConstants.USE_COMPRESS) {
        // 对 http 响应结果开启 gizp 压缩
        pipeline.addLast(new HttpContentCompressor());
    }
    if(CommonConstants.USE_AGGREGATOR) {
        // 将多个HttpRequest组合成一个FullHttpRequest
        pipeline.addLast(new HttpObjectAggregator(CommonConstants.MAX_CONTENT_LENGTH));
    }
}
 

protected ChannelHandler setupHttpChannel(Configuration config, SslContext sslCtx) {

        return new ChannelInitializer<SocketChannel>() {

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

                ch.pipeline().addLast("ssl", new NonSslRedirectHandler(config, sslCtx));
                ch.pipeline().addLast("encoder", new HttpResponseEncoder());
                ch.pipeline().addLast("decoder", new HttpRequestDecoder());
                ch.pipeline().addLast("compressor", new HttpContentCompressor());
                ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
                ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
                ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
                final Configuration.Cors corsCfg = config.getHttp().getCors();
                final CorsConfigBuilder ccb;
                if (corsCfg.isAllowAnyOrigin()) {
                    ccb = CorsConfigBuilder.forAnyOrigin();
                } else {
                    ccb = CorsConfigBuilder.forOrigins(corsCfg.getAllowedOrigins().stream().toArray(String[]::new));
                }
                if (corsCfg.isAllowNullOrigin()) {
                    ccb.allowNullOrigin();
                }
                if (corsCfg.isAllowCredentials()) {
                    ccb.allowCredentials();
                }
                corsCfg.getAllowedMethods().stream().map(HttpMethod::valueOf).forEach(ccb::allowedRequestMethods);
                corsCfg.getAllowedHeaders().forEach(ccb::allowedRequestHeaders);
                CorsConfig cors = ccb.build();
                LOG.trace("Cors configuration: {}", cors);
                ch.pipeline().addLast("cors", new CorsHandler(cors));
                ch.pipeline().addLast("queryDecoder", new qonduit.netty.http.HttpRequestDecoder(config));
                ch.pipeline().addLast("strict", new StrictTransportHandler(config));
                ch.pipeline().addLast("login", new X509LoginRequestHandler(config));
                ch.pipeline().addLast("doLogin", new BasicAuthLoginRequestHandler(config));
                ch.pipeline().addLast("error", new HttpExceptionHandler());
            }
        };
    }
 

@Override
protected void initChannel(final SocketChannel channel) {
    final ChannelPipeline pipeline = channel.pipeline();
    final ChannelHandlerAdapter handler;
    final boolean degzip;
    if (Handlers.isActive("capture")) {
        degzip = true;
        handler = new DefaultResponseLocatorCapturingHandler(api);
    } else if (Handlers.isActive("passthrough")) {
        degzip = false;
        handler = new PassthroughHandler(api);
    } else {
        degzip = true;
        handler = new ServingProxyHandler(api);
    }
    pipeline
            .addLast("logging", new LoggingHandler(LogLevel.valueOf(api.getLogLevel())))
            .addLast("http-decoder", new HttpRequestDecoder());
    if (degzip) {
        pipeline.addLast("gzip-decompressor", new HttpContentDecompressor());
    }
    pipeline
            .addLast("http-encoder", new HttpResponseEncoder())
            .addLast("gzip-compressor", new HttpContentCompressor())
            .addLast("aggregator", new HttpObjectAggregator(Integer.MAX_VALUE))
            .addLast("chunked-writer", new ChunkedWriteHandler())
            .addLast("talend-junit-api-server", handler);
}
 

@Override
public void handlePipeline(ChannelPipeline pipeline) {
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpServerExpectContinueHandler());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new ChunkedWriteHandler());
    pipeline.addLast(new HttpContentCompressor());
    pipeline.addLast(new HttpRequestHandler(sslContextManager));
}
 

protected ChannelHandler setupHttpChannel(BalancerConfiguration balancerConfig, SslContext sslCtx,
        MetricResolver metricResolver, HttpClientPool httpClientPool) {

    return new ChannelInitializer<SocketChannel>() {

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

            ch.pipeline().addLast("ssl", new NonSslRedirectHandler(balancerConfig.getHttp(), sslCtx));
            ch.pipeline().addLast("encoder", new HttpResponseEncoder());
            ch.pipeline().addLast("decoder", new HttpRequestDecoder());
            ch.pipeline().addLast("compressor", new HttpContentCompressor());
            ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
            ch.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
            ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
            ch.pipeline().addLast("queryDecoder", new timely.netty.http.HttpRequestDecoder(
                    balancerConfig.getSecurity(), balancerConfig.getHttp()));
            ch.pipeline().addLast("fileServer", new HttpStaticFileServerHandler().setIgnoreSslHandshakeErrors(
                    balancerConfig.getSecurity().getServerSsl().isUseGeneratedKeypair()));
            ch.pipeline().addLast("login",
                    new X509LoginRequestHandler(balancerConfig.getSecurity(), balancerConfig.getHttp()));
            ch.pipeline().addLast("httpRelay", new HttpRelayHandler(metricResolver, httpClientPool));
            ch.pipeline().addLast("error", new TimelyExceptionHandler().setIgnoreSslHandshakeErrors(
                    balancerConfig.getSecurity().getServerSsl().isUseGeneratedKeypair()));
        }
    };
}
 

protected ChannelHandler setupHttpChannel(GrafanaAuthConfiguration config, SslContext sslCtx,
        HttpClientPool httpClientPool) {

    return new ChannelInitializer<SocketChannel>() {

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

            ch.pipeline().addLast("ssl", new NonSslRedirectHandler(config.getHttp(), sslCtx));
            ch.pipeline().addLast("encoder", new HttpResponseEncoder());
            ch.pipeline().addLast("decoder", new HttpRequestDecoder());
            ch.pipeline().addLast("compressor", new HttpContentCompressor());
            ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
            // high maximum contentLength so that grafana snapshots can be delivered
            // might not be necessary if inbound chunking (while proxying) is handled
            ch.pipeline().addLast("aggregator", new HttpObjectAggregator(2097152));
            ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
            ch.pipeline().addLast("grafanaDecoder",
                    new GrafanaRequestDecoder(config.getSecurity(), config.getHttp()));
            ch.pipeline().addLast("fileServer", new HttpStaticFileServerHandler());
            ch.pipeline().addLast("login", new X509LoginRequestHandler(config.getSecurity(), config.getHttp()));
            ch.pipeline().addLast("httpRelay", new GrafanaRelayHandler(config, httpClientPool));
            ch.pipeline().addLast("error", new TimelyExceptionHandler()
                    .setIgnoreSslHandshakeErrors(config.getSecurity().getServerSsl().isUseGeneratedKeypair()));
        }
    };
}
 

@Override
protected void startUp() throws Exception {
  channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
  EventLoopGroup bossGroup = new NioEventLoopGroup(NUM_BOSS_THREADS,
                                                   new ThreadFactoryBuilder()
                                                     .setDaemon(true).setNameFormat("boss-thread").build());
  EventLoopGroup workerGroup = new NioEventLoopGroup(NUM_WORKER_THREADS,
                                                     new ThreadFactoryBuilder()
                                                       .setDaemon(true).setNameFormat("worker-thread#%d").build());

  bootstrap = new ServerBootstrap()
    .group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<SocketChannel>() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
        channelGroup.add(ch);
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("codec", new HttpServerCodec());
        pipeline.addLast("compressor", new HttpContentCompressor());
        pipeline.addLast("aggregator", new HttpObjectAggregator(MAX_INPUT_SIZE));
        pipeline.addLast("handler", new ReportHandler());
      }
    });

  Channel serverChannel = bootstrap.bind(new InetSocketAddress(host, 0)).sync().channel();
  channelGroup.add(serverChannel);

  bindAddress = (InetSocketAddress) serverChannel.localAddress();
  url = URI.create(String.format("http://%s:%d", host, bindAddress.getPort())).toURL();

  LOG.info("Tracker service started at {}", url);
}
 

@Override
protected void initChannel(SocketChannel ch) throws Exception {
	if ("true".equalsIgnoreCase(Settings.INSTANCE.getProperty("webserver.logging.writelogOfNettyLogger"))) {
		ch.pipeline().addLast("log", new LoggingHandler("lannister.web/server", LogLevel.DEBUG));
	}

	if (useSsl) {
		SslContext sslCtx = SslContextBuilder
				.forServer(Settings.INSTANCE.certChainFile(), Settings.INSTANCE.privateKeyFile()).build();

		logger.debug("SSL Provider : {}", SslContext.defaultServerProvider());

		ch.pipeline().addLast(sslCtx.newHandler(ch.alloc()));
	}

	ch.pipeline().addLast(HttpServerCodec.class.getName(), new HttpServerCodec());
	ch.pipeline().addLast(HttpObjectAggregator.class.getName(), new HttpObjectAggregator(1048576));
	ch.pipeline().addLast(HttpContentCompressor.class.getName(), new HttpContentCompressor());
	ch.pipeline().addLast(HttpRequestRouter.class.getName(), new HttpRequestRouter());

	if (websocketFrameHandlerClass != null) {
		WebsocketFrameHandler wsfh = websocketFrameHandlerClass.newInstance();

		ch.pipeline().addLast(WebSocketServerProtocolHandler.class.getName(), new WebSocketServerProtocolHandler(
				wsfh.websocketPath(), wsfh.subprotocols(), wsfh.allowExtensions(), wsfh.maxFrameSize()));

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

@Override
protected void initChannel(SocketChannel channel) throws Exception {
    logger.debug("Initializing channel from client {} to the server", channel.remoteAddress());

    ChannelPipeline pipeline = channel.pipeline();

    if (sslContext != null) {
        pipeline.addLast(sslContext.newHandler(channel.alloc()));
    }

    pipeline.addLast(new HttpServerCodec(4096, 8192, 65536));
    pipeline.addLast(new HttpContentDecompressor());
    pipeline.addLast(new HttpContentCompressor());
    pipeline.addLast(new ClientToServerChannelHandler(threadPool, route, errorResponseConstructor));
}
 

private void switchToHttp(ChannelHandlerContext ctx) {
    ChannelPipeline p = ctx.pipeline();
    p.addLast("decoder", new HttpRequestDecoder());
    p.addLast("encoder", new HttpResponseEncoder());
    p.addLast("deflater", new HttpContentCompressor());
    p.addLast("handler", new HttpSnoopServerHandler());
    p.remove(this);
}
 

public void run() throws InterruptedException {
    bossGroup = new NioEventLoopGroup(configuration.getThreads());
    workerGroup = new NioEventLoopGroup(configuration.getThreads());

    ServerBootstrap b = new ServerBootstrap();

    b.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 100)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(new HttpRequestDecoder(
          configuration.getMaxInitialLineLength(),
                        configuration.getMaxHeaderSize(),
   configuration.getMaxChunkSize()
        		));
                    p.addLast(new HttpObjectAggregator(MAX_CONTENT_LENGTH));
                    p.addLast(new HttpResponseEncoder());
                    p.addLast(new HttpContentCompressor());
                    p.addLast(new ReaderServerHandler(handlers));
                }

                @Override
                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                    logger.error(cause);
                    super.exceptionCaught(ctx, cause);
                }
            });

    // Start the server.
    b.bind(configuration.getBind(), configuration.getPort()).sync();
}
 

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

  // Uncomment the following line if you want HTTPS
  //SSLEngine engine = SecureChatSslContextFactory.getServerContext().createSSLEngine();
  //engine.setUseClientMode(false);
  //p.addLast("ssl", new SslHandler(engine));

  p.addLast("decoder", new HttpRequestDecoder(16384, 8192, 16384));
  p.addLast("encoder", new HttpResponseEncoder());
  p.addLast("aggregator", new HttpObjectAggregator(maxContentLength));

  p.addLast("chunkedWriter", new ChunkedWriteHandler());
  p.addLast("static", staticFileServerHandler);

  // the compressor is behind the static handler to avoid compression of static files
  // Netty doesn't handle it very well :(
  if (supportZip) {
    p.addLast("compressor", new HttpContentCompressor());
  }

  p.addLast("idleState", new IdleStateHandler(0, 0, idleTimeoutMs, TimeUnit.MILLISECONDS));

  if (corsConfig.isCorsSupportEnabled()) {
    p.addLast("cors", new CorsWrapperHandler(corsConfig));
  }

  p.addLast("handler", new HttpRequestDispatcherHandler(contextPath, dispatcher, staticResolver,
      marshallerRegistry, activeChannels, acceptKeepAlive, requestTimeoutMs, internalErrors, requestTimeoutErrors, notFoundErrors, unexpectedErrors, ioErrors));
}
 

@Override
public void initChannel(Channel ch) throws Exception {
    ch.pipeline().addLast("openChannels", transport.serverOpenChannels);
    ch.pipeline().addLast("read_timeout", new ReadTimeoutHandler(transport.readTimeoutMillis, TimeUnit.MILLISECONDS));
    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()));
    aggregator.setMaxCumulationBufferComponents(transport.maxCompositeBufferComponents);
    ch.pipeline().addLast("aggregator", aggregator);
    if (transport.compression) {
        ch.pipeline().addLast("encoder_compress", new HttpContentCompressor(transport.compressionLevel));
    }
    ch.pipeline().addLast("handler", new MainAndStaticFileHandler(
        nodeName,
        home,
        nodeClient,
        transport.getCorsConfig()
    ));
    pipelineRegistry.registerItems(ch.pipeline(), transport.getCorsConfig());
    if (SETTING_CORS_ENABLED.get(transport.settings())) {
        ch.pipeline().addAfter("encoder", "cors", new Netty4CorsHandler(transport.getCorsConfig()));
    }
}
 

/**
 * Issue the WebSocket upgrade
 *
 * @param exchange The {@link WebSocketHttpExchange} for which the handshake and upgrade should occur.
 */
public final void handshake(final WebSocketHttpExchange exchange, Consumer<ChannelHandlerContext> completeListener) {
    String origin = exchange.getRequestHeader(HttpHeaderNames.ORIGIN);
    if (origin != null) {
        exchange.setResponseHeader(HttpHeaderNames.ORIGIN, origin);
    }
    selectSubprotocol(exchange);
    List<WebSocketServerExtension> extensions = selectExtensions(exchange);
    exchange.setResponseHeader(HttpHeaderNames.SEC_WEB_SOCKET_LOCATION, getWebSocketLocation(exchange));

    final String key = exchange.getRequestHeader(HttpHeaderNames.SEC_WEB_SOCKET_KEY);
    try {
        final String solution = solve(key);
        exchange.setResponseHeader(HttpHeaderNames.SEC_WEB_SOCKET_ACCEPT, solution);
        performUpgrade(exchange);
    } catch (NoSuchAlgorithmException e) {
        exchange.endExchange();
        return;
    }
    handshakeInternal(exchange);
    exchange.upgradeChannel(new Consumer<Object>() {
        @Override
        public void accept(Object c) {
            ChannelHandlerContext context = (ChannelHandlerContext) c;
            WebSocket13FrameDecoder decoder = new WebSocket13FrameDecoder(true, allowExtensions, maxFrameSize, false);
            WebSocket13FrameEncoder encoder = new WebSocket13FrameEncoder(false);
            ChannelPipeline p = context.pipeline();
            if (p.get(HttpObjectAggregator.class) != null) {
                p.remove(HttpObjectAggregator.class);
            }
            if (p.get(HttpContentCompressor.class) != null) {
                p.remove(HttpContentCompressor.class);
            }
            p.addLast("ws-encoder", encoder);
            p.addLast("ws-decoder", decoder);
            for(WebSocketServerExtension extension : extensions) {
                WebSocketExtensionDecoder exdecoder = extension.newExtensionDecoder();
                WebSocketExtensionEncoder exencoder = extension.newExtensionEncoder();
                p.addAfter("ws-decoder", exdecoder.getClass().getName(), exdecoder);
                p.addAfter("ws-encoder", exencoder.getClass().getName(), exencoder);
            }

            completeListener.accept(context);
        }
    });
}