Enable IdleTimeoutHandler only during periods without active HTTP message exchange

Signed-off-by: raccoonback <[email protected]>

Author: raccoonback

reactor-netty-http/src/main/java/reactor/netty/http/server/HttpTrafficHandler.java

@@ -15,17 +15,17 @@
  */
 package reactor.netty.http.server;
 
-import java.net.SocketAddress;
-import java.time.Duration;
-import java.time.ZonedDateTime;
-import java.util.Optional;
-import java.util.Queue;
-import java.util.function.BiFunction;
-import java.util.function.BiPredicate;
+import static io.netty.handler.codec.http.HttpUtil.isContentLengthSet;
+import static io.netty.handler.codec.http.HttpUtil.isKeepAlive;
+import static io.netty.handler.codec.http.HttpUtil.isTransferEncodingChunked;
+import static io.netty.handler.codec.http.HttpUtil.setKeepAlive;
+import static io.netty.handler.codec.http.LastHttpContent.EMPTY_LAST_CONTENT;
+import static reactor.netty.ReactorNetty.format;
 
 import io.netty.channel.ChannelDuplexHandler;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
 import io.netty.channel.ChannelPromise;
 import io.netty.handler.codec.DecoderResult;
 import io.netty.handler.codec.DecoderResultProvider;
@@ -43,28 +43,32 @@
 import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.handler.codec.http.cookie.ServerCookieDecoder;
 import io.netty.handler.codec.http.cookie.ServerCookieEncoder;
+import io.netty.handler.codec.http2.Http2FrameCodec;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.util.ReferenceCountUtil;
+import java.net.SocketAddress;
+import java.time.Duration;
+import java.time.ZonedDateTime;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.function.BiFunction;
+import java.util.function.BiPredicate;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Mono;
 import reactor.netty.Connection;
 import reactor.netty.ConnectionObserver;
 import reactor.netty.ReactorNetty;
 import reactor.netty.channel.ChannelOperations;
+import reactor.netty.http.Http2ConnectionLiveness;
 import reactor.netty.http.Http2SettingsSpec;
+import reactor.netty.http.HttpConnectionImmediateClose;
+import reactor.netty.http.IdleTimeoutHandler;
 import reactor.netty.http.logging.HttpMessageArgProviderFactory;
 import reactor.netty.http.logging.HttpMessageLogFactory;
 import reactor.netty.http.server.compression.HttpCompressionOptionsSpec;
 import reactor.util.annotation.Nullable;
 import reactor.util.concurrent.Queues;
 
-import static io.netty.handler.codec.http.HttpUtil.isContentLengthSet;
-import static io.netty.handler.codec.http.HttpUtil.isKeepAlive;
-import static io.netty.handler.codec.http.HttpUtil.isTransferEncodingChunked;
-import static io.netty.handler.codec.http.HttpUtil.setKeepAlive;
-import static io.netty.handler.codec.http.LastHttpContent.EMPTY_LAST_CONTENT;
-import static reactor.netty.ReactorNetty.format;
-
 /**
  * Replace {@link io.netty.handler.codec.http.HttpServerKeepAliveHandler} with extra
  * handler management.
@@ -157,6 +161,12 @@ public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
 		ctx.read();
 	}
 
+	@Override
+	public void channelActive(ChannelHandlerContext ctx) {
+		setupIdleTimeoutHandler(ctx.pipeline());
+		ctx.fireChannelActive();
+	}
+
 	@Override
 	public void channelRead(ChannelHandlerContext ctx, Object msg) {
 		read = true;
@@ -172,6 +182,10 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
 		if (msg instanceof HttpRequest) {
 			finalizingResponse = false;
 
+			if (idleTimeout != null) {
+				IdleTimeoutHandler.removeIdleTimeoutHandler(ctx.pipeline());
+			}
+
 			final HttpRequest request = (HttpRequest) msg;
 
 			if (H2.equals(request.protocolVersion())) {
@@ -539,6 +553,7 @@ void handleLastHttpContent(Object msg, ChannelPromise promise) {
 			ctx.executor().execute(this);
 		}
 		else {
+			setupIdleTimeoutHandler(ctx.pipeline());
 			ctx.read();
 		}
 	}
@@ -682,6 +697,29 @@ static boolean isMultipart(HttpResponse response) {
 				MULTIPART_PREFIX.length());
 	}
 
+	private void setupIdleTimeoutHandler(ChannelPipeline pipeline) {
+		Http2FrameCodec httpCodec = pipeline.get(Http2FrameCodec.class);
+		if(httpCodec != null) {
+			IdleTimeoutHandler.addIdleTimeoutServerHandler(
+					pipeline,
+					idleTimeout,
+					new Http2ConnectionLiveness(
+							httpCodec,
+							http2SettingsSpec != null ? http2SettingsSpec.pingAckTimeout() : null,
+							http2SettingsSpec != null ? http2SettingsSpec.pingScheduleInterval() : null,
+							http2SettingsSpec != null ? http2SettingsSpec.pingAckDropThreshold() : null
+					)
+			);
+			return;
+		}
+
+		IdleTimeoutHandler.addIdleTimeoutServerHandler(
+				pipeline,
+				idleTimeout,
+				new HttpConnectionImmediateClose()
+		);
+	}
+
 	static final class HttpRequestHolder {
 		final HttpRequest request;
 		final ZonedDateTime timestamp;