HttpServer: Add API for read related timeouts (#2836)

Fixes #2770

docs/asciidoc/http-server.adoc

@@ -685,9 +685,31 @@ This section describes various timeout configuration options that can be used in
 Configuring a proper timeout may improve or solve issues in the communication process.
 The configuration options can be grouped as follows:
 
+* <<http-server-request-timeout>>
 * <<http-server-connection-timeout>>
 * <<http-server-ssl-tls-timeout>>
 
+[[http-server-request-timeout]]
+=== Request Timeout
+The following listing shows all available request timeout configuration options.
+
+* `readTimeout` - the maximum time between each network-level read operation while reading a given request content (resolution: ms)
+* `requestTimeout` - the maximum time for reading a given request content (resolution: ms).
+
+NOTE: It is always a good practice to configure a read/request timeout.
+
+To customize the default settings, you can configure `HttpServer` as follows:
+
+====
+[source,java,indent=0]
+.{examplesdir}/read/timeout/Application.java
+----
+include::{examplesdir}/read/timeout/Application.java[lines=18..36]
+----
+<1> Configures the read timeout to 5 second.
+<2> Configures the request timeout to 30 second.
+====
+
 [[http-server-connection-timeout]]
 === Connection Timeout
 The following listing shows all available connection timeout configuration options.

reactor-netty-core/src/main/java/reactor/netty/NettyPipeline.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2022 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2011-2023 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -121,6 +121,7 @@ public interface NettyPipeline {
 	String ProxyLoggingHandler   = LEFT + "proxyLoggingHandler";
 	String ProxyProtocolDecoder  = LEFT + "proxyProtocolDecoder";
 	String ProxyProtocolReader   = LEFT + "proxyProtocolReader";
+	String ReadTimeoutHandler    = LEFT + "readTimeoutHandler";
 	String ResponseTimeoutHandler = LEFT + "responseTimeoutHandler";
 	String SslHandler            = LEFT + "sslHandler";
 	String SslLoggingHandler     = LEFT + "sslLoggingHandler";

reactor-netty-examples/src/main/java/reactor/netty/examples/documentation/http/server/read/timeout/Application.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2023 VMware, Inc. or its affiliates, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package reactor.netty.examples.documentation.http.server.read.timeout;
+
+import reactor.netty.DisposableServer;
+import reactor.netty.http.server.HttpServer;
+
+import java.time.Duration;
+
+public class Application {
+
+	public static void main(String[] args) {
+		DisposableServer server =
+				HttpServer.create()
+				          .readTimeout(Duration.ofSeconds(5))     //<1>
+				          .requestTimeout(Duration.ofSeconds(30)) //<2>
+				          .handle((request, response) -> request.receive().then())
+				          .bindNow();
+
+		server.onDispose()
+		      .block();
+	}
+}

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

@@ -16,6 +16,7 @@
 package reactor.netty.http.server;
 
 import java.net.SocketAddress;
+import java.time.Duration;
 import java.time.ZonedDateTime;
 import java.util.Optional;
 import java.util.function.BiFunction;
@@ -65,6 +66,8 @@ final class Http2StreamBridgeServerHandler extends ChannelDuplexHandler implemen
 	final ConnectionObserver                                      listener;
 	final BiFunction<? super Mono<Void>, ? super Connection, ? extends Mono<Void>>
 	                                                              mapHandle;
+	final Duration                                                readTimeout;
+	final Duration                                                requestTimeout;
 
 	SocketAddress remoteAddress;
 
@@ -83,7 +86,9 @@ final class Http2StreamBridgeServerHandler extends ChannelDuplexHandler implemen
 			@Nullable BiFunction<ConnectionInfo, HttpRequest, ConnectionInfo> forwardedHeaderHandler,
 			HttpMessageLogFactory httpMessageLogFactory,
 			ConnectionObserver listener,
-			@Nullable BiFunction<? super Mono<Void>, ? super Connection, ? extends Mono<Void>> mapHandle) {
+			@Nullable BiFunction<? super Mono<Void>, ? super Connection, ? extends Mono<Void>> mapHandle,
+			@Nullable Duration readTimeout,
+			@Nullable Duration requestTimeout) {
 		this.compress = compress;
 		this.cookieDecoder = decoder;
 		this.cookieEncoder = encoder;
@@ -92,6 +97,8 @@ final class Http2StreamBridgeServerHandler extends ChannelDuplexHandler implemen
 		this.httpMessageLogFactory = httpMessageLogFactory;
 		this.listener = listener;
 		this.mapHandle = mapHandle;
+		this.readTimeout = readTimeout;
+		this.requestTimeout = requestTimeout;
 	}
 
 	@Override
@@ -135,6 +142,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
 						httpMessageLogFactory,
 						true,
 						mapHandle,
+						readTimeout,
+						requestTimeout,
 						secured,
 						timestamp);
 			}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2022 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2011-2023 VMware, Inc. or its affiliates, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -53,6 +53,7 @@
 import reactor.netty.transport.ServerTransport;
 import reactor.util.Logger;
 import reactor.util.Loggers;
+import reactor.util.annotation.Nullable;
 import reactor.util.context.Context;
 
 import static reactor.netty.ReactorNetty.format;
@@ -739,6 +740,51 @@ public final HttpServer proxyProtocol(ProxyProtocolSupportType proxyProtocolSupp
 		return dup;
 	}
 
+	/**
+	 * Specifies the maximum duration allowed between each network-level read operation while reading a given request
+	 * content (resolution: ms). In other words, {@link io.netty.handler.timeout.ReadTimeoutHandler} is added to the
+	 * channel pipeline after all the request headers are received, and removed from the channel pipeline after the
+	 * content is fully received.
+	 * If the {@code readTimeout} is {@code null}, any previous setting will be removed and no
+	 * {@code readTimeout} will be applied.
+	 * If the {@code readTimeout} is less than {@code 1ms}, then {@code 1ms} will be the
+	 * {@code readTimeout}.
+	 *
+	 * @param readTimeout the maximum duration allowed between each network-level read operation while reading a given
+	 * request content (resolution: ms)
+	 * @return a new {@link HttpServer}
+	 * @since 1.1.9
+	 * @see io.netty.handler.timeout.ReadTimeoutHandler
+	 */
+	public final HttpServer readTimeout(@Nullable Duration readTimeout) {
+		if (Objects.equals(readTimeout, configuration().readTimeout)) {
+			return this;
+		}
+		HttpServer dup = duplicate();
+		dup.configuration().readTimeout = readTimeout;
+		return dup;
+	}
+
+	/**
+	 * Specifies the maximum duration for reading a given request content (resolution: ms).
+	 * If the {@code requestTimeout} is {@code null}, any previous setting will be removed and no
+	 * {@code requestTimeout} will be applied.
+	 * If the {@code requestTimeout} is less than {@code 1ms}, then {@code 1ms} will be the
+	 * {@code requestTimeout}.
+	 *
+	 * @param requestTimeout the maximum duration for reading a given request content (resolution: ms)
+	 * @return a new {@link HttpServer}
+	 * @since 1.1.9
+	 */
+	public final HttpServer requestTimeout(@Nullable Duration requestTimeout) {
+		if (Objects.equals(requestTimeout, configuration().requestTimeout)) {
+			return this;
+		}
+		HttpServer dup = duplicate();
+		dup.configuration().requestTimeout = requestTimeout;
+		return dup;
+	}
+
 	/**
 	 * Define routes for the server through the provided {@link HttpServerRoutes} builder.
 	 *