Java 类io.netty.handler.codec.http.HttpServerCodec 实例源码

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 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 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));
}

@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);
}

@Test
public void addByteDecoderWhenFullReactorPipeline() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .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(NettyPipeline.HttpCodec,
                    NettyPipeline.HttpServerHandler,
                    "decoder$extract",
                    "decoder",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@Test
public void addNonByteDecoderWhenFullReactorPipeline() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler decoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerLast("decoder", decoder);

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

@Test
public void addByteEncoderWhenFullReactorPipeline() throws Exception {

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

    testContext.addHandlerFirst("encoder", encoder);

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

@Test
public void addNonByteEncoderWhenFullReactorPipeline() throws Exception {

    channel.pipeline()
           .addLast(NettyPipeline.HttpCodec, new HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });
    ChannelHandler encoder = new ChannelHandlerAdapter() {
    };

    testContext.addHandlerFirst("encoder", encoder);

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    NettyPipeline.HttpServerHandler,
                    "encoder",
                    NettyPipeline.ReactiveBridge,
                    "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 HttpServerCodec())
           .addLast(NettyPipeline.HttpServerHandler, new ChannelDuplexHandler())
           .addLast(NettyPipeline.ReactiveBridge, new ChannelHandlerAdapter() {
           });

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

    assertEquals(channel.pipeline()
                        .names(),
            Arrays.asList(NettyPipeline.HttpCodec,
                    NettyPipeline.HttpServerHandler,
                    "encoder2",
                    "encoder1",
                    NettyPipeline.ReactiveBridge,
                    "DefaultChannelPipeline$TailContext#0"));
}

@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);
}

protected ChannelInitializer<Channel> createInitializer() {

    return new ChannelInitializer<Channel>() {

        @Override
        protected void initChannel(Channel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();
            p.addLast(new HttpServerCodec() );
            p.addLast(new ChunkedWriteHandler());
            p.addLast(new HttpObjectAggregator(64 * 1024));
            p.addLast(new EchoServerHttpRequestHandler("/ws"));
            p.addLast(new WebSocketServerProtocolHandler("/ws"));
            p.addLast(new EchoServerWSHandler());
        }
    };
}

private void configureHttp1WithUpgrade(ChannelHandlerContext ctx) {
    final ChannelPipeline p = ctx.pipeline();
    final HttpServerCodec http1codec = new HttpServerCodec(
            config.defaultMaxHttp1InitialLineLength(),
            config.defaultMaxHttp1HeaderSize(),
            config.defaultMaxHttp1ChunkSize());

    String baseName = name;
    baseName = addAfter(p, baseName, http1codec);
    baseName = addAfter(p, baseName, new HttpServerUpgradeHandler(
            http1codec,
            protocol -> {
                if (!AsciiString.contentEquals(Http2CodecUtil.HTTP_UPGRADE_PROTOCOL_NAME, protocol)) {
                    return null;
                }

                return new Http2ServerUpgradeCodec(
                        newHttp2ConnectionHandler(p));
            },
            UPGRADE_REQUEST_MAX_LENGTH));

    addAfter(p, baseName, new Http1RequestDecoder(config, ctx.channel(), SCHEME_HTTP));
}

@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));
}

public HttpAPIServer(Holder holder) {
    super(holder.props.getProperty("listen.address"),
            holder.props.getIntProperty("http.port"), holder.transportTypeHolder);

    String adminRootPath = holder.props.getProperty("admin.rootPath", "/admin");

    final HttpAndWebSocketUnificatorHandler httpAndWebSocketUnificatorHandler =
            new HttpAndWebSocketUnificatorHandler(holder, port, adminRootPath);
    final LetsEncryptHandler letsEncryptHandler = new LetsEncryptHandler(holder.sslContextHolder.contentHolder);

    channelInitializer = new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline()
            .addLast("HttpServerCodec", new HttpServerCodec())
            .addLast("HttpServerKeepAlive", new HttpServerKeepAliveHandler())
            .addLast("HttpObjectAggregator", new HttpObjectAggregator(holder.limits.webRequestMaxSize, true))
            .addLast(letsEncryptHandler)
            .addLast("HttpChunkedWrite", new ChunkedWriteHandler())
            .addLast("HttpUrlMapper", new UrlReWriterHandler("/favicon.ico", "/static/favicon.ico"))
            .addLast("HttpStaticFile", new StaticFileHandler(holder.props, new StaticFile("/static"),
                    new StaticFileEdsWith(CSVGenerator.CSV_DIR, ".csv.gz")))
            .addLast("HttpWebSocketUnificator", httpAndWebSocketUnificatorHandler);
        }
    };
}

@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 : instantiateAuthenticationHandler(settings.authentication);
        if (authenticationHandler != null)
            pipeline.addLast(PIPELINE_AUTHENTICATOR, authenticationHandler);
    }

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

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline p = ch.pipeline();
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    if (enableGzip) {
        p.addLast(new HttpContentCompressor());
    }
    p.addLast(new HttpServerCodec(36192 * 2, 36192 * 8, 36192 * 16, false));
    p.addLast(new HttpServerExpectContinueHandler());
    p.addLast(new HttpObjectAggregator(Integer.MAX_VALUE));
    p.addLast(new ChunkedWriteHandler());
    if (enableCors) {
        CorsConfig corsConfig = CorsConfigBuilder.forAnyOrigin().allowNullOrigin().allowCredentials().build();
        p.addLast(new CorsHandler(corsConfig));
    }
    if (null != blade.webSocketPath()) {
        p.addLast(new WebSocketServerProtocolHandler(blade.webSocketPath(), null, true));
        p.addLast(new WebSockerHandler(blade));
    }
    service.scheduleWithFixedDelay(() -> date = new AsciiString(DateKit.gmtDate(LocalDateTime.now())), 1000, 1000, TimeUnit.MILLISECONDS);
    p.addLast(new HttpServerHandler());
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();
    // Enable HTTPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    p.addLast("logger", new LoggingHandler(LogLevel.DEBUG));
    p.addLast(new HttpServerCodec());
    //p.addLast(new HttpRequestDecoder());
    // Uncomment the following line if you don't want to handle HttpChunks.
    //p.addLast(new HttpObjectAggregator(1048576));
    //p.addLast(new HttpResponseEncoder());
    // Remove the following line if you don't want automatic content compression.
    //p.addLast(new HttpContentCompressor());
    p.addLast("aggregator", new HttpObjectAggregator(1048576));
    p.addLast(new HttpHandler(transport, path, connectionListener));
}

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    //Decode  bytes  to  HTTP  requests  /  encode  HTTP requests to bytes.
    pipeline.addLast(new HttpServerCodec());
    //Allows to write a file content.
    pipeline.addLast(new ChunkedWriteHandler());
    //Aggregate decoded HttpRequest / HttpContent / LastHttpContent to FullHttpRequest. This way you will always receive only full Http requests
    pipeline.addLast(new HttpObjectAggregator(64 * 1024));
    //Handle FullHttpRequest which are not send to /ws URI and so serve the index.html page
    pipeline.addLast(new HttpRequestHandler("/ws"));
    //Handle the WebSocket upgrade and Ping/Pong/Close WebSocket frames to be RFC compliant
    pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
    //Handles Text frames and handshake completion events
    pipeline.addLast(new TextWebSocketFrameHandler(group));
}

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

    ChannelPipeline p = ch.pipeline();

    //p.addLast(new HttpResponseEncoder());
    //p.addLast(new HttpRequestDecoder());
    p.addLast(new HttpServerCodec());
    p.addLast(new HttpObjectAggregator(1048576));
    p.addLast(new ChunkedWriteHandler());
    // Remove the following line if you don't want automatic content compression.
    //p.addLast(new HttpContentCompressor());
    p.addLast(new MiddlewareChanelHandler(app));


}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ChannelPipeline pipeline = socketChannel.pipeline();
    pipeline.addLast("timeout", new ReadTimeoutHandler(15));
    pipeline.addLast("codec-http", new HttpServerCodec());
    pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
    pipeline.addLast("handler", new HTTPHandler(plugin));
    pipeline.addLast("websocket", new WebSocketServerProtocolHandler("/server"));
    pipeline.addLast("packet-decoder", new PacketDecoder());
    pipeline.addLast("packet-encoder", new PacketEncoder());
    pipeline.addLast("packet-handler", new ClientHandler(socketChannel, plugin));

    socketChannel.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    plugin.getWebHandler().getChannelGroup().add(socketChannel);
}

@Override
public void initChannel(SocketChannel channel) throws Exception {
    LOGGER.trace("{}: setup", Channels.getHexText(channel));

    // time how long channels live
    channel.closeFuture().addListener(new GenericFutureListener<Future<Void>>() {

        private final Timer.Context lifetimeContext = CHANNEL_LIFETIME_TIMER.time();

        @Override
        public void operationComplete(Future<Void> future) throws Exception {
            lifetimeContext.stop();
        }
    });

    // create the channel pipeline
    channel.pipeline().addLast(
            new IdleTimeoutHandler(0, 0, (int) http.getIdleTimeout(), TimeUnit.MILLISECONDS),
            new HttpServerCodec(HTTP_MAX_INITIAL_LINE_LENGTH, HTTP_MAX_HEADER_SIZE, HTTP_MAX_CHUNK_SIZE, false),
            requestHeaderAssigner,
            new BufferingHttpObjectHandler(),
            new HttpRequestHandler(applicationHandler, baseUri, applicationExecutor, timer),
            finalInboundHandler
            );
}

@Override
public void initChannel(SocketChannel ch) {
  ChannelPipeline p = ch.pipeline();
  p.addLast(new ReadTimeoutHandler(60, TimeUnit.SECONDS));
  if (sslContext != null) {
    p.addLast(sslContext.newHandler(ch.alloc()));
  }
  p.addLast(new HttpContentCompressor(5));
  p.addLast(new HttpServerCodec());
  p.addLast(new HttpObjectAggregator(1048576));
  p.addLast(new ChunkedWriteHandler());
  if (null != corsConfig) {
    p.addLast(new CorsHandler(corsConfig));
  }
  p.addLast(new WebSocketServerCompressionHandler());
  p.addLast(new WebSocketServerProtocolHandler(webSocketPath, null, true));
  p.addLast(new LaputaServerHandler(null != sslContext, requestProcessor));
}

@Override
  protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(new ProxyHandler());
      ch.pipeline().addLast(new HttpServerCodec());
      ch.pipeline().addLast(new HttpObjectAggregator(1024 * 10));
      ch.pipeline().addLast(new AutoCloseHandler());
      ch.pipeline().addLast(new HttpResponseDecorator());
ch.pipeline().addLast("catchall", new HttpCatchAllHandler());

      for (Provider<? extends ChannelHandler> provider : appHandlers) {
          try {
        ChannelHandler handler = provider.get();
        ch.pipeline().addBefore("catchall", handler.getClass().getName(), handler);
    } catch (Exception e) {
        LOG.error(e.getMessage());
        // Push the error back into the pipeline so we can handle gracefully
        // with the HttpCatchAllHandler
        ch.pipeline().fireExceptionCaught(e);
    }
      }
  }

private void switchToHttp(ChannelHandlerContext ctx, ByteBuf msg) {
    ChannelPipeline pipeline = ctx.pipeline();

    addLastIfNotPresent(pipeline, new HttpServerCodec(8192, 8192, 8192));
    addLastIfNotPresent(pipeline, new HttpContentDecompressor());
    addLastIfNotPresent(pipeline, httpContentLengthRemover);
    addLastIfNotPresent(pipeline, new HttpObjectAggregator(Integer.MAX_VALUE));

    if (mockServerLogger.isEnabled(TRACE)) {
        addLastIfNotPresent(pipeline, loggingHandler);
    }
    configurePipeline(ctx, pipeline);
    pipeline.remove(this);

    ctx.channel().attr(LOCAL_HOST_HEADERS).set(getLocalAddresses(ctx));

    // fire message back through pipeline
    ctx.fireChannelRead(msg);
}

@Test
public void shouldReturnBytes() {
    // given
    ChannelHandlerContext mockChannelHandlerContext = mock(ChannelHandlerContext.class);
    ChannelPipeline mockChannelPipeline = mock(ChannelPipeline.class);
    ChannelFuture mockChannelFuture = mock(ChannelFuture.class);

    when(mockChannelHandlerContext.pipeline()).thenReturn(mockChannelPipeline);
    when(mockChannelPipeline.context(HttpServerCodec.class)).thenReturn(mockChannelHandlerContext);
    when(mockChannelHandlerContext.writeAndFlush(any(ByteBuf.class))).thenReturn(mockChannelFuture);

    // when
    new HttpErrorActionHandler().handle(
            error()
                    .withResponseBytes("some_bytes".getBytes()),
            mockChannelHandlerContext
    );

    // then
    verify(mockChannelHandlerContext).pipeline();
    verify(mockChannelPipeline).context(HttpServerCodec.class);
    verify(mockChannelHandlerContext).writeAndFlush(Unpooled.wrappedBuffer("some_bytes".getBytes()));
    verify(mockChannelFuture).awaitUninterruptibly();
}

/**
 * Returns a server-side channel initializer capable of securely receiving
 * and sending HTTP requests and responses
 * <p/>
 * <p>Communications will be encrypted as per the configured SSL context</p>
 *
 * @param handler the handler implementing the business logic.
 * @param sslContext the SSL context which drives the security of the
 * link to the client.
 */
public static final ChannelInitializer<Channel> secureHttpServer(
    final SimpleChannelInboundHandler<HttpRequest> handler,
    final SSLContext sslContext) {
  return new ChannelInitializer<Channel>() {
    @Override
    protected void initChannel(Channel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();
      SSLEngine sslEngine = sslContext.createSSLEngine();
      sslEngine.setUseClientMode(false);
      pipeline.addLast("ssl", new SslHandler(sslEngine));
      pipeline.addLast("httpCodec", new HttpServerCodec());
      pipeline.addLast("aggregator", new HttpObjectAggregator(10 * 1024 * 1024));
      pipeline.addLast("httpServerHandler", handler);
    }
  };
}

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

    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpContentCompressor());
    pipeline.addLast(new HttpObjectAggregator(CommonConstants.MAX_CONTENT_LENGTH));
    pipeline.addLast(new ChunkedWriteHandler());
    pipeline.addLast(new MasterProxyHandler());
}

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

public void start(int port) {
    bossGroup = new NioEventLoopGroup();
    workerGroup = new NioEventLoopGroup();
    try {
      ServerBootstrap b = new ServerBootstrap();
      b.group(bossGroup, workerGroup)
          .channel(NioServerSocketChannel.class)
//          .option(ChannelOption.SO_BACKLOG, 100)
//          .handler(new LoggingHandler(LogLevel.DEBUG))
          .childHandler(new ChannelInitializer<Channel>() {

            @Override
            protected void initChannel(Channel ch) throws Exception {
              ch.pipeline().addLast("httpCodec", new HttpServerCodec());
              ch.pipeline().addLast("serverHandle",
                  new HttpProxyServerHandle(serverConfig, proxyInterceptInitializer, proxyConfig,
                      httpProxyExceptionHandle));
            }
          });
      ChannelFuture f = b
          .bind(port)
          .sync();
      f.channel().closeFuture().sync();
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      bossGroup.shutdownGracefully();
      workerGroup.shutdownGracefully();
    }
  }

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();
    p.addLast(new HttpServerCodec());
    p.addLast(new HttpRequestDecoder());
    p.addLast(new HttpObjectAggregator(20248));
    p.addLast(new VINettyHandler());
    p.addLast(new HttpHelloWorldServerHandler());
}

private void switchToHttp(ChannelHandlerContext ctx) {
    ChannelPipeline p = ctx.pipeline();
    p.addLast(new HttpServerCodec());
    p.addLast(new HttpObjectAggregator(65536));
    p.addLast(new WebSocketServerCompressionHandler());
    p.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, "ws", true));
    p.addLast(new NetoJsonStringToMapWebSocketDecoder());
    p.addLast(new NetoMessageToWebsocketFrameEncoder());
    p.remove(this);

    // 핸들러를 다시 등록 했으므로 이벤트를 전파
    ctx.fireChannelActive();
}

@Override
public void initChannel(final SocketChannel ch) throws Exception {
    ch.pipeline().addLast(new HttpServerCodec(),
            new MqttMessageWebSocketFrameEncoder(),
            new HttpObjectAggregator(65536), httpRequestHandler,
            new WebSocketServerProtocolHandler(websocketUri),
            new MqttMessageWebSocketFrameDecoder(),
            new MqttMessageHandler());
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline p = ch.pipeline();
    p.addLast("codec", new HttpServerCodec(4096, 8192, 8192, false));
    p.addLast("servletInput", new ServletContentHandler(servletContext));
    p.addLast(servletExecutor, "filterChain", requestDispatcherHandler);
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    if (sslContext != null) {
        pipeline.addLast("ssl", sslContext.newHandler(ch.alloc()));
    }
    pipeline.addLast(new HttpServerCodec())
        .addLast(new HttpObjectAggregator(65536))
        .addLast(new ChunkedWriteHandler())
        .addLast(new NettyHttpServerHandler());
}

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