Merge #3453 into 2.0.0-M4

reactor-netty5-http/src/main/java/reactor/netty5/http/HttpDecoderSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2023 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2019-2024 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.
@@ -30,13 +30,15 @@ public abstract class HttpDecoderSpec<T extends HttpDecoderSpec<T>> implements S
 	public static final boolean DEFAULT_VALIDATE_HEADERS                = true;
 	public static final int DEFAULT_INITIAL_BUFFER_SIZE                 = 128;
 	public static final boolean DEFAULT_ALLOW_DUPLICATE_CONTENT_LENGTHS = false;
+	public static final boolean DEFAULT_ALLOW_PARTIAL_CHUNKS = true;
 
 	protected int maxInitialLineLength             = DEFAULT_MAX_INITIAL_LINE_LENGTH;
 	protected int maxHeaderSize                    = DEFAULT_MAX_HEADER_SIZE;
 	protected boolean validateHeaders              = DEFAULT_VALIDATE_HEADERS;
 	protected int initialBufferSize                = DEFAULT_INITIAL_BUFFER_SIZE;
 	protected boolean allowDuplicateContentLengths = DEFAULT_ALLOW_DUPLICATE_CONTENT_LENGTHS;
 	protected int h2cMaxContentLength;
+	protected boolean allowPartialChunks = DEFAULT_ALLOW_PARTIAL_CHUNKS;
 
 	/**
 	 * Configure the maximum length that can be decoded for the HTTP request's initial
@@ -189,6 +191,30 @@ public int h2cMaxContentLength() {
 		return h2cMaxContentLength;
 	}
 
+	/**
+	 * Configure whether chunks can be split into multiple messages, if their chunk size exceeds the size of the
+	 * input buffer. Defaults to {@link #DEFAULT_ALLOW_PARTIAL_CHUNKS}.
+	 *
+	 * @param allowPartialChunks set to {@code false} to only allow sending whole chunks down the pipeline.
+	 * @return this option builder for further configuration
+	 * @since 1.1.23
+	 */
+	public T allowPartialChunks(boolean allowPartialChunks) {
+		this.allowPartialChunks = allowPartialChunks;
+		return get();
+	}
+
+	/**
+	 * Return the configuration whether chunks can be split into multiple messages, if their chunk size
+	 * exceeds the size of the input buffer.
+	 *
+	 * @return the configuration whether to allow duplicate {@code Content-Length} headers
+	 * @since 1.1.23
+	 */
+	public boolean allowPartialChunks() {
+		return allowPartialChunks;
+	}
+
 	@Override
 	public boolean equals(Object o) {
 		if (this == o) {
@@ -202,7 +228,8 @@ public boolean equals(Object o) {
 				validateHeaders == that.validateHeaders &&
 				initialBufferSize == that.initialBufferSize &&
 				allowDuplicateContentLengths == that.allowDuplicateContentLengths &&
-				h2cMaxContentLength == that.h2cMaxContentLength;
+				h2cMaxContentLength == that.h2cMaxContentLength &&
+				allowPartialChunks == that.allowPartialChunks;
 	}
 
 	@Override
@@ -214,6 +241,7 @@ public int hashCode() {
 		result = 31 * result + initialBufferSize;
 		result = 31 * result + Boolean.hashCode(allowDuplicateContentLengths);
 		result = 31 * result + h2cMaxContentLength;
+		result = 31 * result + Boolean.hashCode(allowPartialChunks);
 		return result;
 	}
 }

reactor-netty5-http/src/main/java/reactor/netty5/http/client/HttpClientConfig.java

@@ -635,7 +635,8 @@ static void configureHttp11OrH2CleartextPipeline(
 		             .setMaxHeaderSize(decoder.maxHeaderSize())
 		             .setValidateHeaders(decoder.validateHeaders())
 		             .setInitialBufferSize(decoder.initialBufferSize())
-		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths());
+		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths())
+		             .setAllowPartialChunks(decoder.allowPartialChunks());
 		HttpClientCodec httpClientCodec =
 				new HttpClientCodec(decoderConfig, decoder.failOnMissingResponse, decoder.parseHttpAfterConnectRequest);
 
@@ -695,7 +696,8 @@ static void configureHttp11Pipeline(ChannelPipeline p,
 		             .setMaxHeaderSize(decoder.maxHeaderSize())
 		             .setValidateHeaders(decoder.validateHeaders())
 		             .setInitialBufferSize(decoder.initialBufferSize())
-		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths());
+		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths())
+		             .setAllowPartialChunks(decoder.allowPartialChunks());
 		p.addBefore(NettyPipeline.ReactiveBridge,
 				NettyPipeline.HttpCodec,
 				new HttpClientCodec(decoderConfig, decoder.failOnMissingResponse, decoder.parseHttpAfterConnectRequest));

reactor-netty5-http/src/main/java/reactor/netty5/http/client/HttpResponseDecoderSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2023 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2019-2024 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.
@@ -16,7 +16,6 @@
 package reactor.netty5.http.client;
 
 import reactor.netty5.http.HttpDecoderSpec;
-import reactor.netty5.http.server.HttpRequestDecoderSpec;
 
 /**
  * A configuration builder to fine tune the {@link io.netty5.handler.codec.http.HttpClientCodec}
@@ -32,6 +31,7 @@
  *     <tr><td>{@link #DEFAULT_MAX_INITIAL_LINE_LENGTH}</td><td>4096</td></tr>
  *     <tr><td>{@link #DEFAULT_PARSE_HTTP_AFTER_CONNECT_REQUEST}</td><td>false</td></tr>
  *     <tr><td>{@link #DEFAULT_VALIDATE_HEADERS}</td><td>true</td></tr>
+ *     <tr><td>{@link #DEFAULT_ALLOW_PARTIAL_CHUNKS}</td><td>true</td></tr>
  * </table>
  *
  * @author Violeta Georgieva
@@ -109,7 +109,7 @@ public int hashCode() {
 	}
 
 	/**
-	 * Build a {@link HttpRequestDecoderSpec}.
+	 * Build a {@link HttpResponseDecoderSpec}.
 	 */
 	HttpResponseDecoderSpec build() {
 		HttpResponseDecoderSpec decoder = new HttpResponseDecoderSpec();
@@ -121,6 +121,7 @@ HttpResponseDecoderSpec build() {
 		decoder.failOnMissingResponse = failOnMissingResponse;
 		decoder.parseHttpAfterConnectRequest = parseHttpAfterConnectRequest;
 		decoder.h2cMaxContentLength = h2cMaxContentLength;
+		decoder.allowPartialChunks = allowPartialChunks;
 		return decoder;
 	}
 }

reactor-netty5-http/src/main/java/reactor/netty5/http/server/HttpRequestDecoderSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2022 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2018-2024 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.
@@ -30,6 +30,7 @@
  *     <tr><td>{@link #DEFAULT_MAX_HEADER_SIZE}</td><td>8192</td></tr>
  *     <tr><td>{@link #DEFAULT_MAX_INITIAL_LINE_LENGTH}</td><td>4096</td></tr>
  *     <tr><td>{@link #DEFAULT_VALIDATE_HEADERS}</td><td>true</td></tr>
+ *     <tr><td>{@link #DEFAULT_ALLOW_PARTIAL_CHUNKS}</td><td>true</td></tr>
  * </table>
  *
  * @author Simon Baslé
@@ -64,6 +65,7 @@ HttpRequestDecoderSpec build() {
 		decoder.validateHeaders = validateHeaders;
 		decoder.allowDuplicateContentLengths = allowDuplicateContentLengths;
 		decoder.h2cMaxContentLength = h2cMaxContentLength;
+		decoder.allowPartialChunks = allowPartialChunks;
 		return decoder;
 	}
 }

reactor-netty5-http/src/main/java/reactor/netty5/http/server/HttpServerConfig.java

@@ -612,7 +612,8 @@ static void configureHttp11OrH2CleartextPipeline(ChannelPipeline p,
 		             .setMaxHeaderSize(decoder.maxHeaderSize())
 		             .setValidateHeaders(decoder.validateHeaders())
 		             .setInitialBufferSize(decoder.initialBufferSize())
-		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths());
+		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths())
+		             .setAllowPartialChunks(decoder.allowPartialChunks());
 		HttpServerCodec httpServerCodec =
 				new HttpServerCodec(decoderConfig);
 
@@ -694,7 +695,8 @@ static void configureHttp11Pipeline(ChannelPipeline p,
 		             .setMaxHeaderSize(decoder.maxHeaderSize())
 		             .setValidateHeaders(decoder.validateHeaders())
 		             .setInitialBufferSize(decoder.initialBufferSize())
-		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths());
+		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths())
+		             .setAllowPartialChunks(decoder.allowPartialChunks());
 		p.addBefore(NettyPipeline.ReactiveBridge,
 		            NettyPipeline.HttpCodec,
 		            new HttpServerCodec(decoderConfig))

reactor-netty5-http/src/test/java/reactor/netty5/http/HttpDecoderSpecTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2019-2022 VMware, Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2019-2024 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.
@@ -42,6 +42,7 @@ void maxInitialLineLength() {
 		checkDefaultValidateHeaders(conf);
 		checkDefaultInitialBufferSize(conf);
 		checkDefaultAllowDuplicateContentLengths(conf);
+		checkDefaultAllowPartialChunks(conf);
 	}
 
 	@Test
@@ -69,6 +70,7 @@ void maxHeaderSize() {
 		checkDefaultValidateHeaders(conf);
 		checkDefaultInitialBufferSize(conf);
 		checkDefaultAllowDuplicateContentLengths(conf);
+		checkDefaultAllowPartialChunks(conf);
 	}
 
 	@Test
@@ -96,6 +98,7 @@ void validateHeaders() {
 		checkDefaultMaxHeaderSize(conf);
 		checkDefaultInitialBufferSize(conf);
 		checkDefaultAllowDuplicateContentLengths(conf);
+		checkDefaultAllowPartialChunks(conf);
 	}
 
 	@Test
@@ -110,6 +113,7 @@ void initialBufferSize() {
 		checkDefaultMaxHeaderSize(conf);
 		checkDefaultValidateHeaders(conf);
 		checkDefaultAllowDuplicateContentLengths(conf);
+		checkDefaultAllowPartialChunks(conf);
 	}
 
 	@Test
@@ -137,6 +141,22 @@ void allowDuplicateContentLengths() {
 		checkDefaultMaxHeaderSize(conf);
 		checkDefaultValidateHeaders(conf);
 		checkDefaultInitialBufferSize(conf);
+		checkDefaultAllowPartialChunks(conf);
+	}
+
+	@Test
+	void allowPartialChunks() {
+		checkDefaultAllowPartialChunks(conf);
+
+		conf.allowPartialChunks(false);
+
+		assertThat(conf.allowPartialChunks()).as("allow partial chunks").isFalse();
+
+		checkDefaultMaxInitialLineLength(conf);
+		checkDefaultMaxHeaderSize(conf);
+		checkDefaultValidateHeaders(conf);
+		checkDefaultInitialBufferSize(conf);
+		checkDefaultAllowDuplicateContentLengths(conf);
 	}
 
 	public static void checkDefaultMaxInitialLineLength(HttpDecoderSpec<?> conf) {
@@ -169,6 +189,12 @@ public static void checkDefaultAllowDuplicateContentLengths(HttpDecoderSpec<?> c
 				.isFalse();
 	}
 
+	public static void checkDefaultAllowPartialChunks(HttpDecoderSpec<?> conf) {
+		assertThat(conf.allowPartialChunks()).as("default allow partial chunks")
+				.isEqualTo(HttpDecoderSpec.DEFAULT_ALLOW_PARTIAL_CHUNKS)
+				.isTrue();
+	}
+
 	private static final class HttpDecoderSpecImpl extends HttpDecoderSpec<HttpDecoderSpecImpl> {
 
 		@Override

reactor-netty5-http/src/test/java/reactor/netty5/http/client/HttpClientTest.java

@@ -1731,6 +1731,7 @@ void httpClientResponseConfigInjectAttributes() {
 		AtomicReference<Channel> channelRef = new AtomicReference<>();
 		AtomicBoolean validate = new AtomicBoolean();
 		AtomicBoolean allowDuplicateContentLengths = new AtomicBoolean();
+		AtomicBoolean allowPartialChunks = new AtomicBoolean(true);
 		disposableServer =
 				createServer()
 				          .handle((req, resp) -> req.receive()
@@ -1744,7 +1745,8 @@ void httpClientResponseConfigInjectAttributes() {
 		                                       .initialBufferSize(10)
 		                                       .failOnMissingResponse(true)
 		                                       .parseHttpAfterConnectRequest(true)
-		                                       .allowDuplicateContentLengths(true))
+		                                       .allowDuplicateContentLengths(true)
+		                                       .allowPartialChunks(false))
 		        .doOnConnected(c -> {
 		                    channelRef.set(c.channel());
 		                    HttpClientCodec codec = c.channel()
@@ -1753,6 +1755,7 @@ void httpClientResponseConfigInjectAttributes() {
 		                    HttpObjectDecoder decoder = (HttpObjectDecoder) getValueReflection(codec, "inboundHandler", 1);
 		                    validate.set((Boolean) getValueReflection(decoder, "validateHeaders", 2));
 		                    allowDuplicateContentLengths.set((Boolean) getValueReflection(decoder, "allowDuplicateContentLengths", 2));
+		                    allowPartialChunks.set((Boolean) getValueReflection(decoder, "allowPartialChunks", 2));
 		                })
 		        .post()
 		        .uri("/")
@@ -1765,6 +1768,7 @@ void httpClientResponseConfigInjectAttributes() {
 		assertThat(channelRef.get()).isNotNull();
 		assertThat(validate).as("validate headers").isFalse();
 		assertThat(allowDuplicateContentLengths).as("allow duplicate Content-Length").isTrue();
+		assertThat(allowPartialChunks).as("allow partial chunks").isFalse();
 	}
 
 	private Object getValueReflection(Object obj, String fieldName, int superLevel) {

reactor-netty5-http/src/test/java/reactor/netty5/http/server/HttpServerTests.java

@@ -969,13 +969,15 @@ void httpServerRequestConfigInjectAttributes() {
 		AtomicReference<Channel> channelRef = new AtomicReference<>();
 		AtomicBoolean validate = new AtomicBoolean();
 		AtomicBoolean allowDuplicateContentLengths = new AtomicBoolean();
+		AtomicBoolean allowPartialChunks = new AtomicBoolean(true);
 		disposableServer =
 				createServer()
 				          .httpRequestDecoder(opt -> opt.maxInitialLineLength(123)
 				                                        .maxHeaderSize(456)
 				                                        .validateHeaders(false)
 				                                        .initialBufferSize(10)
-				                                        .allowDuplicateContentLengths(true))
+				                                        .allowDuplicateContentLengths(true)
+				                                        .allowPartialChunks(false))
 				          .handle((req, resp) -> req.receive().then(resp.sendNotFound()))
 				          .doOnConnection(c -> {
 				                      channelRef.set(c.channel());
@@ -985,6 +987,7 @@ void httpServerRequestConfigInjectAttributes() {
 				                      HttpObjectDecoder decoder = (HttpObjectDecoder) getValueReflection(codec, "inboundHandler", 1);
 				                      validate.set((Boolean) getValueReflection(decoder, "validateHeaders", 2));
 				                      allowDuplicateContentLengths.set((Boolean) getValueReflection(decoder, "allowDuplicateContentLengths", 2));
+				                      allowPartialChunks.set((Boolean) getValueReflection(decoder, "allowPartialChunks", 2));
 				                  })
 				          .bindNow();
 
@@ -1000,6 +1003,7 @@ void httpServerRequestConfigInjectAttributes() {
 		assertThat(channelRef.get()).isNotNull();
 		assertThat(validate).as("validate headers").isFalse();
 		assertThat(allowDuplicateContentLengths).as("allow duplicate Content-Length").isTrue();
+		assertThat(allowPartialChunks).as("allow partial chunks").isFalse();
 	}
 
 	private Object getValueReflection(Object obj, String fieldName, int superLevel) {