Java 类io.netty.channel.ChannelPipeline 实例源码

@Override
protected void initChannel(Channel ch) throws Exception {
    RPCChannelHandler channelHandler = 
            new RPCChannelHandler(syncManager, rpcService);

    IdleStateHandler idleHandler = 
            new IdleStateHandler(5, 10, 0);
    ReadTimeoutHandler readTimeoutHandler = 
            new ReadTimeoutHandler(30);

    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("idle", idleHandler);
    pipeline.addLast("timeout", readTimeoutHandler);
    pipeline.addLast("handshaketimeout",
                     new HandshakeTimeoutHandler(channelHandler, timer, 10));

    pipeline.addLast("syncMessageDecoder",
                     new SyncMessageDecoder(maxFrameSize));
    pipeline.addLast("syncMessageEncoder",
                     new SyncMessageEncoder());

    pipeline.addLast("handler", channelHandler);
}

/**
 * Creates a handler for interacting with the switch channel
 *
 * @param controller
 *            the controller
 * @param newConnectionListener
 *            the class that listens for new OF connections (switchManager)
 * @param pipeline
 *            the channel pipeline
 * @param threadPool
 *            the thread pool
 * @param idleTimer
 *            the hash wheeled timer used to send idle messages (echo).
 *            passed to constructor to modify in case of aux connection.
 * @param debugCounters
 */
OFChannelHandler(@Nonnull IOFSwitchManager switchManager,
        @Nonnull INewOFConnectionListener newConnectionListener,
        @Nonnull ChannelPipeline pipeline,
        @Nonnull IDebugCounterService debugCounters,
        @Nonnull Timer timer,
        @Nonnull List<U32> ofBitmaps,
        @Nonnull OFFactory defaultFactory) {

    Preconditions.checkNotNull(switchManager, "switchManager");
    Preconditions.checkNotNull(newConnectionListener, "connectionOpenedListener");
    Preconditions.checkNotNull(pipeline, "pipeline");
    Preconditions.checkNotNull(timer, "timer");
    Preconditions.checkNotNull(debugCounters, "debugCounters");

    this.pipeline = pipeline;
    this.debugCounters = debugCounters;
    this.newConnectionListener = newConnectionListener;
    this.counters = switchManager.getCounters();
    this.state = new InitState();
    this.timer = timer;
    this.ofBitmaps = ofBitmaps;
    this.factory = defaultFactory;

    log.debug("constructor on OFChannelHandler {}", String.format("%08x", System.identityHashCode(this)));
}

@Override
public void handleMessage(ChannelHandlerContext ctx, WorldLoginRequest msg) {
    ClientMessage message = evaluateLogin(msg);
    if (message != ClientMessage.SUCCESSFUL) {
        ctx.write(new WorldLoginResponse(message));
        return;
    }

    Player player = new Player(ctx.channel());
    ctx.write(new WorldLoginResponse(player, message, msg.getIsaacPair()));

    ChannelPipeline pipeline = ctx.pipeline();
    pipeline.remove("login.encoder");

    // this isnt set automatically.
    pipeline.addAfter("world.decoder", "game.encoder", new GamePacketEncoder(msg.getIsaacPair().getEncoderRandom()));
    pipeline.replace("world.decoder", "game.decoder", new GamePacketDecoder(player, msg.getIsaacPair().getDecoderRandom()));

    player.init(msg.getDisplayInformation());
}

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

@Override
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(1048576));
    p.addLast(new HttpResponseEncoder());
    // Remove the following line if you don't want automatic content compression.
    //p.addLast(new HttpContentCompressor());
    p.addLast(new MockingFCMServerHandler());
}

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

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

    // handle new connections and idle timeouts
    p.addLast("auth", authHandler);

    // break each data chunk by newlines and split out metrics
    p.addLast("line", new GraphiteMetricDecoder(maxLength));

    // batch up metrics and store
    p.addLast("metrics", new MetricHandler(store));
}

private void unregisterChannelHandler() {
    if (serverChannelHandler == null)
        return;

    for (Channel serverChannel : serverChannels) {
        final ChannelPipeline pipeline = serverChannel.pipeline();

        // Remove channel handler
        serverChannel.eventLoop().execute(new Runnable() {

            @Override
            public void run() {
                try {
                    pipeline.remove(serverChannelHandler);
                } catch (NoSuchElementException e) {
                    // That's fine
                }
            }

        });
    }
}

private void unregisterChannelHandler() {
    if (serverChannelHandler == null)
        return;

    for (Channel serverChannel : serverChannels) {
        final ChannelPipeline pipeline = serverChannel.pipeline();

        // Remove channel handler
        serverChannel.eventLoop().execute(new Runnable() {

            @Override
            public void run() {
                try {
                    pipeline.remove(serverChannelHandler);
                } catch (NoSuchElementException e) {
                    // That's fine
                }
            }

        });
    }
}

@Override
protected void encode(ChannelHandlerContext ctx, ConnectionResponse msg, ByteBuf out) throws Exception {
    ChannelPipeline pipeline = ctx.pipeline();

    switch (msg.getType()) {
    case HANDSHAKE_CONNECTION:
        pipeline.addAfter("decoder", "handshake.encoder", new HandshakeEncoder());
        pipeline.replace("decoder", "handshake.decoder", new HandshakeDecoder());
        break;
    case LOGIN_CONNECTION:
        out.writeByte(ClientMessage.SUCCESSFUL_CONNECTION.getId());
        pipeline.addAfter("decoder", "login.encoder", new LoginEncoder());
        pipeline.replace("decoder", "login.decoder", new LoginDecoder());
        break;
    }

    pipeline.remove(this);
}

@Override
  public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    SslHandler sslHandler = null;
    if (sslHandlerProvider != null) {
      sslHandler = sslHandlerProvider.getSslHandler();
      pipeline.addLast(sslHandler);
    }
    pipeline.addLast("decoder", new MqttDecoder(MAX_PAYLOAD_SIZE));
    pipeline.addLast("encoder", MqttEncoder.INSTANCE);

    MqttTransportHandler handler = new MqttTransportHandler(msgProducer, deviceService, authService, assetService,
        assetAuthService, relationService, sslHandler);
    pipeline.addLast(handler);

//    ch.closeFuture().addListener(handler);

  }

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
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    super.channelActive(ctx);
    logger.info("通道重置");
    List<ChannelPipeline> channelPipelines = tcpMediator.getSendingMsgRepo().getChannelPipelines();
    channelPipelines.add(ctx.pipeline());
    channelPipelines.removeIf(channel -> {
        i++;
        if (channel == null || !channel.channel().isActive()) {
            logger.info("「" + i + "」" + "通道失效");
            return true;
        } else {
            logger.info("「" + i + "」" + "通道有效");
            return false;
        }
    });
    i = 0;
    logger.info("通道数量：" + channelPipelines.size());
}

@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 void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    // 根据服务端协议，选择解码器
    Constants.ProtocolType type = transportConfig.getProvider().getProtocolType();
    switch (type) {
        case jsf:
            pipeline.addLast(new JSFEncoder());
            pipeline.addLast(new JSFDecoder(transportConfig.getPayload()));
            break;
        case dubbo:
            pipeline.addLast(new DubboEncoder());
            pipeline.addLast(new DubboDecoder(transportConfig.getPayload()));
            break;
        default:
            throw new InitErrorException("Unsupported client protocol type : " + type.name());
    }
    pipeline.addLast(Constants.CLIENT_CHANNELHANDLE_NAME, clientChannelHandler);
}

void initiateProtocolOrSsl(String username, String password, String database, Handler<? super CommandResponse<Connection>> completionHandler) {
  ChannelPipeline pipeline = socket.channelHandlerContext().pipeline();
  if (ssl) {
    Future<Void> upgradeFuture = Future.future();
    upgradeFuture.setHandler(ar -> {
      if (ar.succeeded()) {
        initiateProtocol(username, password, database, completionHandler);
      } else {
        Throwable cause = ar.cause();
        if (cause instanceof DecoderException) {
          DecoderException err = (DecoderException) cause;
          cause = err.getCause();
        }
        completionHandler.handle(CommandResponse.failure(cause));
      }
    });
    pipeline.addBefore("handler", "initiate-ssl-handler", new InitiateSslHandler(this, upgradeFuture));
  } else {
    initiateProtocol(username, password, database, completionHandler);
  }
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    final ChannelPipeline pipeline = ch.pipeline();
    NettyPipelineInit.serializePipeline(serializeProtocolEnum, pipeline);
    pipeline.addLast("timeout",
            new IdleStateHandler(nettyConfig.getHeartTime(), nettyConfig.getHeartTime(), nettyConfig.getHeartTime(), TimeUnit.SECONDS));
    pipeline.addLast(nettyServerMessageHandler);
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    final ChannelPipeline pipeline = socketChannel.pipeline();
    NettyPipelineInit.serializePipeline(serializeProtocolEnum, pipeline);
    pipeline.addLast("timeout", new IdleStateHandler(txConfig.getHeartTime(), txConfig.getHeartTime(), txConfig.getHeartTime(), TimeUnit.SECONDS));
    pipeline.addLast(nettyClientMessageHandler);

}

@Override
public void initChannel(SocketChannel channel) throws Exception {
    ClientCrypto clientCrypto = new ClientCrypto(NetworkServer.SERVER_KEY);
    ServerCrypto serverCrypto = new ServerCrypto();

    clientCrypto.setServer(serverCrypto);
    serverCrypto.setClient(clientCrypto);

    ChannelPipeline pipeline = channel.pipeline();
    pipeline.addLast(new PacketDecoder(serverCrypto));
    pipeline.addLast(new PacketEncoder(serverCrypto));
    pipeline.addLast(new PlayerHandler(server));
}

@Override public void prepare(final Benchmark benchmark) {
  this.concurrencyLevel = benchmark.concurrencyLevel;
  this.targetBacklog = benchmark.targetBacklog;

  ChannelInitializer<SocketChannel> channelInitializer = new ChannelInitializer<SocketChannel>() {
    @Override public void initChannel(SocketChannel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();

      if (benchmark.tls) {
        SslClient sslClient = SslClient.localhost();
        SSLEngine engine = sslClient.sslContext.createSSLEngine();
        engine.setUseClientMode(true);
        pipeline.addLast("ssl", new SslHandler(engine));
      }

      pipeline.addLast("codec", new HttpClientCodec());
      pipeline.addLast("inflater", new HttpContentDecompressor());
      pipeline.addLast("handler", new HttpChannel(channel));
    }
  };

  bootstrap = new Bootstrap();
  bootstrap.group(new NioEventLoopGroup(concurrencyLevel))
      .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
      .channel(NioSocketChannel.class)
      .handler(channelInitializer);
}

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

    pipeline.addLast("framer", new LengthFieldBasedFrameDecoder(400 * 1024, 0, 4, -4, 0));
    pipeline.addLast("decoder", new PacketDecoder());
    pipeline.addLast("encoder", new PacketEncoder());
    pipeline.addLast(new LoggingHandler(LogLevel.DEBUG));
    pipeline.addLast("handler", new MessageServerHandler(handlerManager));
}

@Override
protected void channelRead0(ChannelHandlerContext ctx, NetworkMessage msg) throws Exception {
    NetworkMessageListener<NetworkMessage> listener = NetworkMessageRepository.getNetworkListener(msg);

    listener.handleMessage(ctx, msg);

    ChannelPipeline pipeline = ctx.pipeline();
    ChannelHandler handler = msg.getHandler();

    if (pipeline.context(handler) != null) {

        // flush for specific handler.
        pipeline.context(handler).flush();
    }
}

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

    ChannelPipeline p = ch.pipeline();

    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    p.addLast(new HttpClientCodec());
}

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

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

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

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

}

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

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

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

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

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

@Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        logger.info("******", pipeline.toString());
        pipeline.addLast("http-codec", new HttpServerCodec());
        pipeline.addLast("aggregator", new HttpObjectAggregator(65536)); // Http消息组装  
        pipeline.addLast("http-chunked", new ChunkedWriteHandler()); // WebSocket通信支持  
        pipeline.addLast("framer", new LengthFieldBasedFrameDecoder(400 * 1024, 0, 4, -4, 0));
//        pipeline.addLast("decoder", new PacketDecoder());
//        pipeline.addLast("encoder", new PacketEncoder());
        pipeline.addLast(new LoggingHandler(LogLevel.DEBUG));
        pipeline.addLast("handler", new MessageSocketServerHandler(handlerManager));
    }

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

@SideOnly(Side.CLIENT)
private static void addClientHandlers()
{
    ChannelPipeline pipeline = channelPair.get(Side.CLIENT).pipeline();
    String targetName = channelPair.get(Side.CLIENT).findChannelHandlerNameForType(FMLRuntimeCodec.class);
    pipeline.addAfter(targetName, "GuiHandler", new OpenGuiHandler());
    pipeline.addAfter(targetName, "EntitySpawnHandler", new EntitySpawnHandler());
}

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
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ProtobufAdapter adapter = new ProtobufAdapter(config);

    ChannelPipeline pipeline = socketChannel.pipeline();
    pipeline.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());
    pipeline.addLast("decoder", adapter.getDecoder());
    pipeline.addLast("frameEncoder", new ProtobufVarint32LengthFieldPrepender());
    pipeline.addLast("encoder", adapter.getEncoder());
    pipeline.addLast("handler", new TcpServerHandler(config));
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("encoder", new MqttMessageEncoder());
    pipeline.addLast("decoder", new MqttMessageDecoder());
    pipeline.addLast("handler", new MqttMessageHandler());
}

@PostConstruct
public void start() throws Exception {

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

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

@PostConstruct
public void start() throws Exception {


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

                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast("logHandler", new LoggingHandler());
                    pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                    pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
                    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                    pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
                    pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));

                    pipeline.addLast("handler", deviceChannelInboundHandler);
                }
            })

            .option(ChannelOption.SO_BACKLOG, 1000000)  //连接队列深度
            .option(ChannelOption.TCP_NODELAY, true)   //设置 no_delay
            .option(ChannelOption.SO_SNDBUF, 2048).option(ChannelOption.SO_RCVBUF, 1024)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.SO_REUSEADDR, true)
            .childOption(ChannelOption.SO_SNDBUF, 2048).childOption(ChannelOption.SO_RCVBUF, 1024)
            .childOption(ChannelOption.SO_LINGER, 0);

    bootstrap.bind(goPushNodeServerConfig.getDevicePort()).sync();
    log.info("device server start successful! listening port: {}", goPushNodeServerConfig.getDevicePort());
}

@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
 protected void initChannel(SocketChannel ch) throws Exception {
     ChannelPipeline pipeline = ch.pipeline();

     // SSL的安全链接
     if (ServerConfig.isSsl()) {
         SSLContext sslcontext = SSLContext.getInstance("TLS");
         KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509");
         KeyStore ks = KeyStore.getInstance("JKS");
         String keyStorePath = ServerConfig.getKeyStorePath();
         String keyStorePassword = ServerConfig.getKeyStorePassword();
         ks.load(new FileInputStream(keyStorePath), keyStorePassword.toCharArray());
         String keyPassword = ServerConfig.getKeyPassword();
         kmf.init(ks, keyPassword.toCharArray());
         sslcontext.init(kmf.getKeyManagers(), null, null);
         SSLEngine sslEngine = sslcontext.createSSLEngine();
         sslEngine.setUseClientMode(false);
         sslEngine.setNeedClientAuth(false);
         /**
          * 务必放在第一位
          */
         pipeline.addLast(new SslHandler(sslEngine));
         logger.info("initChannel: addLast SslHandler");
         /**
          * Generates a temporary self-signed certificate for testing purposes.
          */
/*SelfSignedCertificate ssc = new SelfSignedCertificate();
SslContext sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
//SslContext sslCtx = SslContext.newServerContext(ssc.certificate(), ssc.privateKey());
if (sslCtx != null) {
    pipeline.addLast(sslCtx.newHandler(ch.alloc()));
}*/
     }
     // Register HTTP handler chain.
     this.appendHttpPipeline(pipeline);
 }

@Override
public void operationComplete(ChannelFuture future) throws Exception {
    connectComplete = true;

    if (future.isSuccess()) {
        if (future.channel().isOpen()) {
            if (trace) {
                log.trace("Channel connect future completed successfully [to={}]", id);
            }
        } else {
            // Channel was disconnect()'ed while we were connecting.
            becomeDisconnected();
        }
    } else if (firstError == null) {
        if (trace) {
            log.trace("Notifying on connect future failure [to={}]", id, future.cause());
        }

        firstError = NettyErrorUtils.unwrap(future.cause());

        ChannelPipeline pipeline = future.channel().pipeline();

        if (pipeline.names().contains(NettyClientStateHandler.class.getName())) {
            pipeline.fireExceptionCaught(firstError);
        } else {
            becomeDisconnected();
        }
    }
}

public NettyServerClient(InetSocketAddress remoteAddress, InetSocketAddress localAddress, boolean ssl, int hbInterval,
    int hbLossThreshold, boolean hbDisabled, Map<String, HandlerRegistration> handlers, EventLoopGroup coreEventLoopGroup) {
    this.remoteAddress = remoteAddress;
    this.localAddress = localAddress;
    this.ssl = ssl;
    this.hbInterval = hbInterval;
    this.hbLossThreshold = hbLossThreshold;
    this.hbDisabled = hbDisabled;
    this.handlers = handlers;
    this.coreEventLoopGroup = coreEventLoopGroup;

    writeListener = future -> {
        if (future.isSuccess()) {
            // Notify metrics on successful operation.
            if (metrics != null) {
                metrics.onMessageDequeue();

                metrics.onMessageSent();
            }
        } else {
            ChannelPipeline pipeline = future.channel().pipeline();

            // Notify on error (only if pipeline is not empty).
            if (pipeline.last() != null) {
                future.channel().pipeline().fireExceptionCaught(future.cause());
            }

            // Notify metrics on failed operation.
            if (metrics != null) {
                metrics.onMessageDequeue();

                metrics.onMessageSendError();
            }
        }
    };

    hbFlushListener = future -> hbFlushed = true;
}

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

    ChannelPipeline pipeline = socketChannel.pipeline();
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new ChunkedWriteHandler());
    pipeline.addLast(new DefaultHttpServerHandler(ahsc));
}

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

    Address upstreamAddress = getAddress();
    SocketAddress address = new InetSocketAddress(upstreamAddress.getHost(), upstreamAddress.getPort());
    pipeline.addFirst(HANDLER_NAME, new SSocksConnectHandler(channel.newPromise(), address, cipher, password));
}

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

    Address upstreamAddress = getAddress();
    SocketAddress address = new InetSocketAddress(upstreamAddress.getHost(), upstreamAddress.getPort());
    pipeline.addFirst(HANDLER_NAME, new Socks5ProxyHandler(address));
}