Support allowPartialChunks in the HttpDecoderSpec

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

reactor-netty-http/src/main/java/reactor/netty/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.
@@ -37,6 +37,7 @@ 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;
@@ -45,6 +46,7 @@ public abstract class HttpDecoderSpec<T extends HttpDecoderSpec<T>> implements S
 	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
@@ -225,6 +227,29 @@ 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
+	 */
+	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) {
@@ -240,7 +265,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
@@ -253,6 +279,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-netty-http/src/main/java/reactor/netty/http/client/HttpClientConfig.java

@@ -653,7 +653,8 @@ static void configureHttp11OrH2CleartextPipeline(
 		             .setMaxChunkSize(decoder.maxChunkSize())
 		             .setValidateHeaders(decoder.validateHeaders())
 		             .setInitialBufferSize(decoder.initialBufferSize())
-		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths());
+		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths())
+		             .setAllowPartialChunks(decoder.allowPartialChunks());
 		HttpClientCodec httpClientCodec =
 				new HttpClientCodec(decoderConfig, decoder.failOnMissingResponse, decoder.parseHttpAfterConnectRequest);
 
@@ -715,7 +716,8 @@ static void configureHttp11Pipeline(ChannelPipeline p,
 		             .setMaxChunkSize(decoder.maxChunkSize())
 		             .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-netty-http/src/main/java/reactor/netty/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.netty.http.client;
 
 import reactor.netty.http.HttpDecoderSpec;
-import reactor.netty.http.server.HttpRequestDecoderSpec;
 
 /**
  * A configuration builder to fine tune the {@link io.netty.handler.codec.http.HttpClientCodec}
@@ -33,6 +32,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
@@ -110,7 +110,7 @@ public int hashCode() {
 	}
 
 	/**
-	 * Build a {@link HttpRequestDecoderSpec}.
+	 * Build a {@link HttpResponseDecoderSpec}.
 	 */
 	HttpResponseDecoderSpec build() {
 		HttpResponseDecoderSpec decoder = new HttpResponseDecoderSpec();
@@ -123,6 +123,7 @@ HttpResponseDecoderSpec build() {
 		decoder.failOnMissingResponse = failOnMissingResponse;
 		decoder.parseHttpAfterConnectRequest = parseHttpAfterConnectRequest;
 		decoder.h2cMaxContentLength = h2cMaxContentLength;
+		decoder.allowPartialChunks = allowPartialChunks;
 		return decoder;
 	}
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018-2021 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.
@@ -31,6 +31,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é
@@ -66,6 +67,7 @@ HttpRequestDecoderSpec build() {
 		decoder.validateHeaders = validateHeaders;
 		decoder.allowDuplicateContentLengths = allowDuplicateContentLengths;
 		decoder.h2cMaxContentLength = h2cMaxContentLength;
+		decoder.allowPartialChunks = allowPartialChunks;
 		return decoder;
 	}
 }

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

@@ -650,7 +650,8 @@ static void configureHttp11OrH2CleartextPipeline(ChannelPipeline p,
 		             .setMaxChunkSize(decoder.maxChunkSize())
 		             .setValidateHeaders(decoder.validateHeaders())
 		             .setInitialBufferSize(decoder.initialBufferSize())
-		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths());
+		             .setAllowDuplicateContentLengths(decoder.allowDuplicateContentLengths())
+		             .setAllowPartialChunks(decoder.allowPartialChunks());
 		HttpServerCodec httpServerCodec =
 				new HttpServerCodec(decoderConfig);
 
@@ -736,7 +737,8 @@ static void configureHttp11Pipeline(ChannelPipeline p,
 		             .setMaxChunkSize(decoder.maxChunkSize())
 		             .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-netty-http/src/test/java/reactor/netty/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.
@@ -43,6 +43,7 @@ void maxInitialLineLength() {
 		checkDefaultValidateHeaders(conf);
 		checkDefaultInitialBufferSize(conf);
 		checkDefaultAllowDuplicateContentLengths(conf);
+		checkDefaultAllowPartialChunks(conf);
 	}
 
 	@Test
@@ -71,6 +72,7 @@ void maxHeaderSize() {
 		checkDefaultValidateHeaders(conf);
 		checkDefaultInitialBufferSize(conf);
 		checkDefaultAllowDuplicateContentLengths(conf);
+		checkDefaultAllowPartialChunks(conf);
 	}
 
 	@Test
@@ -100,6 +102,7 @@ void maxChunkSize() {
 		checkDefaultValidateHeaders(conf);
 		checkDefaultInitialBufferSize(conf);
 		checkDefaultAllowDuplicateContentLengths(conf);
+		checkDefaultAllowPartialChunks(conf);
 	}
 
 	@Test
@@ -129,6 +132,7 @@ void validateHeaders() {
 		checkDefaultMaxChunkSize(conf);
 		checkDefaultInitialBufferSize(conf);
 		checkDefaultAllowDuplicateContentLengths(conf);
+		checkDefaultAllowPartialChunks(conf);
 	}
 
 	@Test
@@ -144,6 +148,7 @@ void initialBufferSize() {
 		checkDefaultMaxChunkSize(conf);
 		checkDefaultValidateHeaders(conf);
 		checkDefaultAllowDuplicateContentLengths(conf);
+		checkDefaultAllowPartialChunks(conf);
 	}
 
 	@Test
@@ -172,6 +177,23 @@ void allowDuplicateContentLengths() {
 		checkDefaultMaxChunkSize(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);
+		checkDefaultMaxChunkSize(conf);
+		checkDefaultValidateHeaders(conf);
+		checkDefaultInitialBufferSize(conf);
+		checkDefaultAllowDuplicateContentLengths(conf);
 	}
 
 	public static void checkDefaultMaxInitialLineLength(HttpDecoderSpec<?> conf) {
@@ -211,6 +233,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-netty-http/src/test/java/reactor/netty/http/client/HttpClientTest.java

@@ -1701,6 +1701,7 @@ void httpClientResponseConfigInjectAttributes() {
 		AtomicBoolean validate = new AtomicBoolean();
 		AtomicInteger chunkSize = new AtomicInteger();
 		AtomicBoolean allowDuplicateContentLengths = new AtomicBoolean();
+		AtomicBoolean allowPartialChunks = new AtomicBoolean();
 		disposableServer =
 				createServer()
 				          .handle((req, resp) -> req.receive()
@@ -1715,7 +1716,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()
@@ -1725,6 +1727,7 @@ void httpClientResponseConfigInjectAttributes() {
 		                    chunkSize.set((Integer) getValueReflection(decoder, "maxChunkSize", 2));
 		                    validate.set((Boolean) getValueReflection(decoder, "validateHeaders", 2));
 		                    allowDuplicateContentLengths.set((Boolean) getValueReflection(decoder, "allowDuplicateContentLengths", 2));
+		                    allowPartialChunks.set((Boolean) getValueReflection(decoder, "allowPartialChunks", 2));
 		                })
 		        .post()
 		        .uri("/")
@@ -1738,6 +1741,7 @@ void httpClientResponseConfigInjectAttributes() {
 		assertThat(chunkSize).as("line length").hasValue(789);
 		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-netty-http/src/test/java/reactor/netty/http/server/HttpServerTests.java

@@ -975,14 +975,16 @@ void httpServerRequestConfigInjectAttributes() {
 		AtomicBoolean validate = new AtomicBoolean();
 		AtomicInteger chunkSize = new AtomicInteger();
 		AtomicBoolean allowDuplicateContentLengths = new AtomicBoolean();
+		AtomicBoolean allowPartialChunks = new AtomicBoolean();
 		disposableServer =
 				createServer()
 				          .httpRequestDecoder(opt -> opt.maxInitialLineLength(123)
 				                                        .maxHeaderSize(456)
 				                                        .maxChunkSize(789)
 				                                        .validateHeaders(false)
 				                                        .initialBufferSize(10)
-				                                        .allowDuplicateContentLengths(true))
+				                                        .allowDuplicateContentLengths(true)
+				                                        .allowPartialChunks(false))
 				          .handle((req, resp) -> req.receive().then(resp.sendNotFound()))
 				          .doOnConnection(c -> {
 				                      channelRef.set(c.channel());
@@ -1009,6 +1011,7 @@ void httpServerRequestConfigInjectAttributes() {
 		assertThat(chunkSize).as("line length").hasValue(789);
 		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) {