import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
@Override
protected void initChannel(io.netty.channel.socket.SocketChannel channel) throws Exception {
    ChannelPipeline pipeline = channel.pipeline();

    // 连接管理
    pipeline.addLast(ConnectionManageHandler.NAME, new ConnectionManageHandler(3000));

    // 空闲超时
    pipeline.addLast(new IdleStateHandler(10, 10, 0));
    pipeline.addLast(new IdleStateEventHandler());

    // 读写超时
    pipeline.addLast(new ReadTimeoutHandler(socksProperties.getReadTimeoutMillis(), TimeUnit.MILLISECONDS));
    pipeline.addLast(new WriteTimeoutHandler(socksProperties.getWriteTimeoutMillis(), TimeUnit.MILLISECONDS));

    // netty log
    //pipeline.addLast(new LoggingHandler());

    // 负责将输出的 Socks5Message 转为 ByteBuf
    pipeline.addLast(Socks5ServerEncoder.DEFAULT);

    // init
    pipeline.addLast(Socks5InitialRequestDecoder.class.getName(), new Socks5InitialRequestDecoder());
    pipeline.addLast(Socks5InitialRequestHandler.class.getName(), socks5InitialRequestHandler);

    // auth
    if (socks5PasswordAuthRequestHandler != null) {
        pipeline.addLast(Socks5PasswordAuthRequestDecoder.class.getName(), new Socks5PasswordAuthRequestDecoder());
        pipeline.addLast(Socks5PasswordAuthRequestHandler.class.getName(), socks5PasswordAuthRequestHandler);
    }

    // connection
    pipeline.addLast(Socks5CommandRequestDecoder.class.getName(), new Socks5CommandRequestDecoder());
    pipeline.addLast(Socks5CommandRequestHandler.class.getName(), socks5CommandRequestHandler);
}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
private void makeRequest(HttpRequest request) {
    log.debug("Writing request: {}", request);
    channel.pipeline().addFirst(new WriteTimeoutHandler(context.configuration().writeTimeout()));
    channel.writeAndFlush(new StreamedRequest(request, context.sdkRequestProvider(), channel))
           .addListener(wireCall -> {
               ChannelUtils.removeIfExists(channel.pipeline(), WriteTimeoutHandler.class);
               if (wireCall.isSuccess()) {
                   channel.pipeline().addFirst(new ReadTimeoutHandler(context.configuration().readTimeout()));
                   // Auto-read is turned off so trigger an explicit read to give control to HttpStreamsClientHandler
                   channel.read();
               } else {
                   handleFailure(() -> "Failed to make request to " + endpoint(), wireCall.cause());
               }
           });
}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
@Override
public void channelReleased(Channel ch) throws Exception {
    // Remove any existing handlers from the pipeline from the previous request.
    ChannelUtils.removeIfExists(ch.pipeline(),
                                HttpStreamsClientHandler.class,
                                ResponseHandler.class,
                                ReadTimeoutHandler.class,
                                WriteTimeoutHandler.class);
}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
@Override
protected void initChannel(SocketChannel channel) throws Exception {

    BackendInfo backendInfo = Apex.getBalancingStrategy()
            .selectBackend(channel.remoteAddress().getHostName(), channel.remoteAddress().getPort());

    if (backendInfo == null) {
        // Gracefully close the channel
        channel.close();

        logger.error("Unable to select a backend server. All down?");
        return;
    }

    channel.pipeline()
            .addLast(new ReadTimeoutHandler(readTimeout))
            .addLast(new WriteTimeoutHandler(writeTimeout));

    GlobalTrafficShapingHandler trafficShapingHandler = Apex.getInstance().getTrafficShapingHandler();
    if (trafficShapingHandler != null) {
        channel.pipeline().addLast(trafficShapingHandler);
    }

    channel.pipeline().addLast(new SocketUpstreamHandler(backendInfo));

    // Keep track of connections per second
    if (connectionsPerSecondTask != null) {
        connectionsPerSecondTask.inc();
    }

    logger.debug("Connected [{}] <-> [{}:{} ({})]", channel.remoteAddress(), backendInfo.getHost(), backendInfo.getPort(), backendInfo.getName());
}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
public ChannelHandler getChangeHandler(HandlerType handlerType,
                                       ClientSessionConfiguration sessionConfiguration,
                                       SessionCallbackHandler clientSession,
                                       Transcoder transcoder) {
    switch (handlerType) {
        case MESSAGE_LOGGER:
            return MESSAGE_LOGGER;
        case SESSION_WRAPPER:
            return new SessionWrapperHandler(clientSession);
        case WRITE_TIMEOUT_HANDLER:
            return new WriteTimeoutHandler(sessionConfiguration.getWriteTimeout(),
                                           TimeUnit.MILLISECONDS);
        case MESSAGE_DECODER:
            return new MessageDecoder(transcoder);
        case MESSAGE_ENCODER:
            return new MessageEncoder(transcoder);
        case LENGTH_FRAME_DECODER:
            return new LengthFieldBasedFrameDecoder(MAXIMUM_MESSAGE_BYTE_SIZE,
                                                    MESSAGE_FIELD_OFFSET,
                                                    LENGTH_FIELD_SIZE,
                                                    0,
                                                    LENGTH_FIELD_SIZE);
        case LENGTH_FRAME_ENCODER:
            return new LengthFieldPrepender(LENGTH_FIELD_SIZE, false);
        default:
            throw new IllegalArgumentException("Invalid handler type");
    }
}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
@Test
public void testCreateWriteTimeoutHandler() throws Exception {
    assertTrue("Not correct class",
               channelHandlerProvider.getChangeHandler(HandlerType.WRITE_TIMEOUT_HANDLER,
                                                       mock(ClientSessionConfiguration.class),
                                                       mock(SessionCallbackHandler.class),
                                                       mock(Transcoder.class))
                       instanceof WriteTimeoutHandler);
}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
@Override
protected void initChannel(SocketChannel ch) throws Exception {
	ch.pipeline().addLast("PushMessageDecoder-" + ch.hashCode(),
			new DDPushMessageDecoder());
	ch.pipeline().addLast("processPushTask-" + ch.hashCode(),
			new PushTaskHandler(
					NettyPushListener.this));
	ch.pipeline().addLast("WritTimeout-" + ch.hashCode(),
			new WriteTimeoutHandler(sockTimeoutSeconds));
	ch.pipeline().addLast(new PushResponseHandler());

}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline p = ch.pipeline();
    p.addLast(new ReadTimeoutHandler(readTimeout, TimeUnit.MILLISECONDS));
    p.addLast(new WriteTimeoutHandler(writeTimeout, TimeUnit.MILLISECONDS));
    p.addLast(serverHandler);
}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
public void refreshWriteTimeoutHandler(Channel channel) {
    if (channel != null) {
        if (this.writeTimeout <= 0) {
            if (channel.pipeline().get("writeTimeout") != null) {
                channel.pipeline().remove("writeTimeout");
            }
        } else if (channel.pipeline().get("writeTimeout") == null) {
            channel.pipeline().addFirst("writeTimeout", new WriteTimeoutHandler(this.writeTimeout));
        } else {
            channel.pipeline().replace("writeTimeout", "writeTimeout", new WriteTimeoutHandler(this.writeTimeout));
        }
    }
}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
protected void refreshWriteTimeoutHandler(Channel channel) {
    if(channel != null) {
        if(this.writeTimeout <= 0) {
            if(channel.pipeline().get("writeTimeout") != null) {
                channel.pipeline().remove("writeTimeout");
            }
        } else {
            if(channel.pipeline().get("writeTimeout") == null) {
                channel.pipeline().addFirst("writeTimeout", new WriteTimeoutHandler(this.writeTimeout));
            } else {
                channel.pipeline().replace("writeTimeout", "writeTimeout", new WriteTimeoutHandler(this.writeTimeout));
            }
        }
    }
}
 

import io.netty.handler.timeout.WriteTimeoutHandler; //导入依赖的package包/类
@Override
public void initChannel( Channel channel )
{
    ChannelPipeline pipeline = channel.pipeline();

    // Connection Events
    String remoteHostString = ( (InetSocketAddress) channel.remoteAddress() ).getHostString();
    app.events().emit( new ConnectionEvent.Opened( remoteHostString ) );
    channel.closeFuture().addListener(
        future -> app.events().emit( new ConnectionEvent.Closed( remoteHostString ) )
    );

    if( app.config().bool( WERVAL_HTTP_LOG_LOWLEVEL_ENABLED ) )
    {
        // Log Netty Bytes
        LogLevel level = LogLevel.valueOf(
            app.config().string( WERVAL_HTTP_LOG_LOWLEVEL_LEVEL ).toUpperCase( US )
        );
        pipeline.addLast( "byte-logging", new LoggingHandler( "io.werval.server.netty.LowLevelLogger", level ) );
    }

    // Read/Write Timeout
    long readTimeout = app.config().seconds( WERVAL_HTTP_TIMEOUT_READ );
    long writeTimeout = app.config().seconds( WERVAL_HTTP_TIMEOUT_WRITE );
    pipeline.addLast( "read-timeout", new ReadTimeoutHandler( readTimeout, SECONDS ) );
    pipeline.addLast( "write-timeout", new WriteTimeoutHandler( writeTimeout, SECONDS ) );

    // HTTP Decoding / Encoding
    // HTTP decoders always generates multiple message objects per a single HTTP message:
    //
    //  1       * HttpRequest / HttpResponse
    //  0 - n   * HttpContent
    //  1       * LastHttpContent
    //
    // or a single FullHttpRequest if a handler ask for it
    pipeline.addLast( "http-codec", new HttpServerCodec() );

    // GZip decompression support
    pipeline.addLast( "http-decompressor", new HttpContentDecompressor() );

    // Allow to send chunked data
    pipeline.addLast( "chunked-write-handler", new ChunkedWriteHandler() );

    // Protocol Switching Handler
    pipeline.addLast( "subprotocol-switcher", new SubProtocolSwitchHandler( allChannels, app, devSpi ) );
}