Expose maxInMemorySize via CodecConfigurer

Centralized maxInMemorySize exposed via CodecConfigurer along with
ability to plug in an instance of MultipartHttpMessageWrite.

Closes gh-23884

spring-core/src/test/java/org/springframework/core/codec/ResourceRegionEncoderTests.java

@@ -19,7 +19,6 @@ package org.springframework.core.codec;
 import java.util.Collections;
 import java.util.function.Consumer;
 
-import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Subscription;
 import reactor.core.publisher.BaseSubscriber;
@@ -33,7 +32,6 @@ import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.AbstractLeakCheckingTests;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
-import org.springframework.core.io.buffer.LeakAwareDataBufferFactory;
 import org.springframework.core.io.buffer.support.DataBufferTestUtils;
 import org.springframework.core.io.support.ResourceRegion;
 import org.springframework.util.MimeType;

spring-web/src/main/java/org/springframework/http/codec/CodecConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -143,6 +143,20 @@ public interface CodecConfigurer {
 		 */
 		void jaxb2Encoder(Encoder<?> encoder);
 
+		/**
+		 * Configure a limit on the number of bytes that can be buffered whenever
+		 * the input stream needs to be aggregated. This can be a result of
+		 * decoding to a single {@code DataBuffer},
+		 * {@link java.nio.ByteBuffer ByteBuffer}, {@code byte[]},
+		 * {@link org.springframework.core.io.Resource Resource}, {@code String}, etc.
+		 * It can also occur when splitting the input stream, e.g. delimited text,
+		 * in which case the limit applies to data buffered between delimiters.
+		 * <p>By default this is not set, in which case individual codec defaults
+		 * apply. All codecs are limited to 256K by default.
+		 * @param byteCount the max number of bytes to buffer, or -1 for unlimited
+		 * @sine 5.1.11
+		 */
+		void maxInMemorySize(int byteCount);
 		/**
 		 * Whether to log form data at DEBUG level, and headers at TRACE level.
 		 * Both may contain sensitive information.

spring-web/src/main/java/org/springframework/http/codec/ServerCodecConfigurer.java

@@ -76,6 +76,20 @@ public interface ServerCodecConfigurer extends CodecConfigurer {
 	 */
 	interface ServerDefaultCodecs extends DefaultCodecs {
 
+		/**
+		 * Configure the {@code HttpMessageReader} to use for multipart requests.
+		 * <p>By default, if
+		 * <a href="https://github.com/synchronoss/nio-multipart">Synchronoss NIO Multipart</a>
+		 * is present, this is set to
+		 * {@link org.springframework.http.codec.multipart.MultipartHttpMessageReader
+		 * MultipartHttpMessageReader} created with an instance of
+		 * {@link org.springframework.http.codec.multipart.SynchronossPartHttpMessageReader
+		 * SynchronossPartHttpMessageReader}.
+		 * @param reader the message reader to use for multipart requests.
+		 * @since 5.1.11
+		 */
+		void multipartReader(HttpMessageReader<?> reader);
+
 		/**
 		 * Configure the {@code Encoder} to use for Server-Sent Events.
 		 * <p>By default if this is not set, and Jackson is available, the

spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartHttpMessageReader.java

@@ -65,6 +65,14 @@ public class MultipartHttpMessageReader extends LoggingCodecSupport
 	}
 
 
+	/**
+	 * Return the configured parts reader.
+	 * @since 5.1.11
+	 */
+	public HttpMessageReader<Part> getPartReader() {
+		return this.partReader;
+	}
+
 	@Override
 	public List<MediaType> getReadableMediaTypes() {
 		return Collections.singletonList(MediaType.MULTIPART_FORM_DATA);

spring-web/src/main/java/org/springframework/http/codec/protobuf/ProtobufDecoder.java

@@ -74,7 +74,7 @@ import org.springframework.util.MimeType;
 public class ProtobufDecoder extends ProtobufCodecSupport implements Decoder<Message> {
 
 	/** The default max size for aggregating messages. */
-	protected static final int DEFAULT_MESSAGE_MAX_SIZE = 64 * 1024;
+	protected static final int DEFAULT_MESSAGE_MAX_SIZE = 256 * 1024;
 
 	private static final ConcurrentMap<Class<?>, Method> methodCache = new ConcurrentReferenceHashMap<>();
 
@@ -102,10 +102,23 @@ public class ProtobufDecoder extends ProtobufCodecSupport implements Decoder<Mes
 	}
 
 
+	/**
+	 * The max size allowed per message.
+	 * <p>By default, this is set to 256K.
+	 * @param maxMessageSize the max size per message, or -1 for unlimited
+	 */
 	public void setMaxMessageSize(int maxMessageSize) {
 		this.maxMessageSize = maxMessageSize;
 	}
 
+	/**
+	 * Return the {@link #setMaxMessageSize configured} message size limit.
+	 * @since 5.1.11
+	 */
+	public int getMaxMessageSize() {
+		return this.maxMessageSize;
+	}
+
 
 	@Override
 	public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType) {
@@ -205,7 +218,7 @@ public class ProtobufDecoder extends ProtobufCodecSupport implements Decoder<Mes
 						if (!readMessageSize(input)) {
 							return messages;
 						}
-						if (this.messageBytesToRead > this.maxMessageSize) {
+						if (this.maxMessageSize > 0 && this.messageBytesToRead > this.maxMessageSize) {
 							throw new DataBufferLimitException(
 									"The number of bytes to read for message " +
 											"(" + this.messageBytesToRead + ") exceeds " +

spring-web/src/main/java/org/springframework/http/codec/support/BaseDefaultCodecs.java

@@ -20,6 +20,7 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
+import org.springframework.core.codec.AbstractDataBufferDecoder;
 import org.springframework.core.codec.ByteArrayDecoder;
 import org.springframework.core.codec.ByteArrayEncoder;
 import org.springframework.core.codec.ByteBufferDecoder;
@@ -29,6 +30,7 @@ import org.springframework.core.codec.DataBufferDecoder;
 import org.springframework.core.codec.DataBufferEncoder;
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.Encoder;
+import org.springframework.core.codec.ResourceDecoder;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.http.codec.CodecConfigurer;
 import org.springframework.http.codec.DecoderHttpMessageReader;
@@ -38,6 +40,7 @@ import org.springframework.http.codec.HttpMessageReader;
 import org.springframework.http.codec.HttpMessageWriter;
 import org.springframework.http.codec.ResourceHttpMessageReader;
 import org.springframework.http.codec.ResourceHttpMessageWriter;
+import org.springframework.http.codec.json.AbstractJackson2Decoder;
 import org.springframework.http.codec.json.Jackson2JsonDecoder;
 import org.springframework.http.codec.json.Jackson2JsonEncoder;
 import org.springframework.http.codec.json.Jackson2SmileDecoder;
@@ -95,6 +98,9 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs {
 	@Nullable
 	private Encoder<?> jaxb2Encoder;
 
+	@Nullable
+	private Integer maxInMemorySize;
+
 	private boolean enableLoggingRequestDetails = false;
 
 	private boolean registerDefaults = true;
@@ -130,6 +136,16 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs {
 		this.jaxb2Encoder = encoder;
 	}
 
+	@Override
+	public void maxInMemorySize(int byteCount) {
+		this.maxInMemorySize = byteCount;
+	}
+
+	@Nullable
+	protected Integer maxInMemorySize() {
+		return this.maxInMemorySize;
+	}
+
 	@Override
 	public void enableLoggingRequestDetails(boolean enable) {
 		this.enableLoggingRequestDetails = enable;
@@ -155,17 +171,20 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs {
 			return Collections.emptyList();
 		}
 		List<HttpMessageReader<?>> readers = new ArrayList<>();
-		readers.add(new DecoderHttpMessageReader<>(new ByteArrayDecoder()));
-		readers.add(new DecoderHttpMessageReader<>(new ByteBufferDecoder()));
-		readers.add(new DecoderHttpMessageReader<>(new DataBufferDecoder()));
-		readers.add(new ResourceHttpMessageReader());
-		readers.add(new DecoderHttpMessageReader<>(StringDecoder.textPlainOnly()));
+		readers.add(new DecoderHttpMessageReader<>(init(new ByteArrayDecoder())));
+		readers.add(new DecoderHttpMessageReader<>(init(new ByteBufferDecoder())));
+		readers.add(new DecoderHttpMessageReader<>(init(new DataBufferDecoder())));
+		readers.add(new ResourceHttpMessageReader(init(new ResourceDecoder())));
+		readers.add(new DecoderHttpMessageReader<>(init(StringDecoder.textPlainOnly())));
 		if (protobufPresent) {
-			Decoder<?> decoder = this.protobufDecoder != null ? this.protobufDecoder : new ProtobufDecoder();
+			Decoder<?> decoder = this.protobufDecoder != null ? this.protobufDecoder : init(new ProtobufDecoder());
 			readers.add(new DecoderHttpMessageReader<>(decoder));
 		}
 
 		FormHttpMessageReader formReader = new FormHttpMessageReader();
+		if (this.maxInMemorySize != null) {
+			formReader.setMaxInMemorySize(this.maxInMemorySize);
+		}
 		formReader.setEnableLoggingRequestDetails(this.enableLoggingRequestDetails);
 		readers.add(formReader);
 
@@ -174,6 +193,28 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs {
 		return readers;
 	}
 
+	private <T extends Decoder<?>> T init(T decoder) {
+		if (this.maxInMemorySize != null) {
+			if (decoder instanceof AbstractDataBufferDecoder) {
+				((AbstractDataBufferDecoder<?>) decoder).setMaxInMemorySize(this.maxInMemorySize);
+			}
+			if (decoder instanceof ProtobufDecoder) {
+				((ProtobufDecoder) decoder).setMaxMessageSize(this.maxInMemorySize);
+			}
+			if (jackson2Present) {
+				if (decoder instanceof AbstractJackson2Decoder) {
+					((AbstractJackson2Decoder) decoder).setMaxInMemorySize(this.maxInMemorySize);
+				}
+			}
+			if (jaxb2Present) {
+				if (decoder instanceof Jaxb2XmlDecoder) {
+					((Jaxb2XmlDecoder) decoder).setMaxInMemorySize(this.maxInMemorySize);
+				}
+			}
+		}
+		return decoder;
+	}
+
 	/**
 	 * Hook for client or server specific typed readers.
 	 */
@@ -189,13 +230,13 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs {
 		}
 		List<HttpMessageReader<?>> readers = new ArrayList<>();
 		if (jackson2Present) {
-			readers.add(new DecoderHttpMessageReader<>(getJackson2JsonDecoder()));
+			readers.add(new DecoderHttpMessageReader<>(init(getJackson2JsonDecoder())));
 		}
 		if (jackson2SmilePresent) {
-			readers.add(new DecoderHttpMessageReader<>(new Jackson2SmileDecoder()));
+			readers.add(new DecoderHttpMessageReader<>(init(new Jackson2SmileDecoder())));
 		}
 		if (jaxb2Present) {
-			Decoder<?> decoder = this.jaxb2Decoder != null ? this.jaxb2Decoder : new Jaxb2XmlDecoder();
+			Decoder<?> decoder = this.jaxb2Decoder != null ? this.jaxb2Decoder : init(new Jaxb2XmlDecoder());
 			readers.add(new DecoderHttpMessageReader<>(decoder));
 		}
 		extendObjectReaders(readers);
@@ -216,7 +257,7 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs {
 			return Collections.emptyList();
 		}
 		List<HttpMessageReader<?>> result = new ArrayList<>();
-		result.add(new DecoderHttpMessageReader<>(StringDecoder.allMimeTypes()));
+		result.add(new DecoderHttpMessageReader<>(init(StringDecoder.allMimeTypes())));
 		return result;
 	}
 

spring-web/src/main/java/org/springframework/http/codec/support/ServerDefaultCodecsImpl.java

@@ -39,10 +39,18 @@ class ServerDefaultCodecsImpl extends BaseDefaultCodecs implements ServerCodecCo
 					DefaultServerCodecConfigurer.class.getClassLoader());
 
 
+	@Nullable
+	private HttpMessageReader<?> multipartReader;
+
 	@Nullable
 	private Encoder<?> sseEncoder;
 
 
+	@Override
+	public void multipartReader(HttpMessageReader<?> reader) {
+		this.multipartReader = reader;
+	}
+
 	@Override
 	public void serverSentEventEncoder(Encoder<?> encoder) {
 		this.sseEncoder = encoder;
@@ -51,10 +59,18 @@ class ServerDefaultCodecsImpl extends BaseDefaultCodecs implements ServerCodecCo
 
 	@Override
 	protected void extendTypedReaders(List<HttpMessageReader<?>> typedReaders) {
+		if (this.multipartReader != null) {
+			typedReaders.add(this.multipartReader);
+			return;
+		}
 		if (synchronossMultipartPresent) {
 			boolean enable = isEnableLoggingRequestDetails();
 
 			SynchronossPartHttpMessageReader partReader = new SynchronossPartHttpMessageReader();
+			Integer size = maxInMemorySize();
+			if (size != null) {
+				partReader.setMaxInMemorySize(size);
+			}
 			partReader.setEnableLoggingRequestDetails(enable);
 			typedReaders.add(partReader);
 

spring-web/src/test/java/org/springframework/http/codec/support/ServerCodecConfigurerTests.java

@@ -36,6 +36,7 @@ import org.springframework.core.codec.DataBufferDecoder;
 import org.springframework.core.codec.DataBufferEncoder;
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.Encoder;
+import org.springframework.core.codec.ResourceDecoder;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.MediaType;
@@ -124,13 +125,45 @@ public class ServerCodecConfigurerTests {
 				.filter(e -> e == encoder).orElse(null)).isSameAs(encoder);
 	}
 
+	@Test
+	public void maxInMemorySize() {
+		int size = 99;
+		this.configurer.defaultCodecs().maxInMemorySize(size);
+		List<HttpMessageReader<?>> readers = this.configurer.getReaders();
+		assertThat(readers.size()).isEqualTo(13);
+		assertThat(((ByteArrayDecoder) getNextDecoder(readers)).getMaxInMemorySize()).isEqualTo(size);
+		assertThat(((ByteBufferDecoder) getNextDecoder(readers)).getMaxInMemorySize()).isEqualTo(size);
+		assertThat(((DataBufferDecoder) getNextDecoder(readers)).getMaxInMemorySize()).isEqualTo(size);
+
+		ResourceHttpMessageReader resourceReader = (ResourceHttpMessageReader) nextReader(readers);
+		ResourceDecoder decoder = (ResourceDecoder) resourceReader.getDecoder();
+		assertThat(decoder.getMaxInMemorySize()).isEqualTo(size);
+
+		assertThat(((StringDecoder) getNextDecoder(readers)).getMaxInMemorySize()).isEqualTo(size);
+		assertThat(((ProtobufDecoder) getNextDecoder(readers)).getMaxMessageSize()).isEqualTo(size);
+		assertThat(((FormHttpMessageReader) nextReader(readers)).getMaxInMemorySize()).isEqualTo(size);
+		assertThat(((SynchronossPartHttpMessageReader) nextReader(readers)).getMaxInMemorySize()).isEqualTo(size);
+
+		MultipartHttpMessageReader multipartReader = (MultipartHttpMessageReader) nextReader(readers);
+		SynchronossPartHttpMessageReader reader = (SynchronossPartHttpMessageReader) multipartReader.getPartReader();
+		assertThat((reader).getMaxInMemorySize()).isEqualTo(size);
+
+		assertThat(((Jackson2JsonDecoder) getNextDecoder(readers)).getMaxInMemorySize()).isEqualTo(size);
+		assertThat(((Jackson2SmileDecoder) getNextDecoder(readers)).getMaxInMemorySize()).isEqualTo(size);
+		assertThat(((Jaxb2XmlDecoder) getNextDecoder(readers)).getMaxInMemorySize()).isEqualTo(size);
+		assertThat(((StringDecoder) getNextDecoder(readers)).getMaxInMemorySize()).isEqualTo(size);
+	}
 
 	private Decoder<?> getNextDecoder(List<HttpMessageReader<?>> readers) {
-		HttpMessageReader<?> reader = readers.get(this.index.getAndIncrement());
+		HttpMessageReader<?> reader = nextReader(readers);
 		assertThat(reader.getClass()).isEqualTo(DecoderHttpMessageReader.class);
 		return ((DecoderHttpMessageReader<?>) reader).getDecoder();
 	}
 
+	private HttpMessageReader<?> nextReader(List<HttpMessageReader<?>> readers) {
+		return readers.get(this.index.getAndIncrement());
+	}
+
 	private Encoder<?> getNextEncoder(List<HttpMessageWriter<?>> writers) {
 		HttpMessageWriter<?> writer = writers.get(this.index.getAndIncrement());
 		assertThat(writer.getClass()).isEqualTo(EncoderHttpMessageWriter.class);

src/docs/asciidoc/web/webflux.adoc

@@ -818,6 +818,33 @@ for repeated, map-like access to parts, or otherwise rely on the
 `SynchronossPartHttpMessageReader` for a one-time access to `Flux<Part>`.
 
 
+[[webflux-codecs-limits]]
+==== Limits
+
+`Decoder` and `HttpMessageReader` implementations that buffer some or all of the input
+stream can be configured with a limit on the maximum number of bytes to buffer in memory.
+In some cases buffering occurs because input is aggregated and represented as a single
+object, e.g. controller method with `@RequestBody byte[]`, `x-www-form-urlencoded` data,
+and so on. Buffering can also occurs with streaming, when splitting the input stream,
+e.g. delimited text, a stream of JSON objects, and so on. For those streaming cases, the
+limit applies to the number of bytes associted with one object in the stream.
+
+To configure buffer sizes, you can check if a given `Decoder` or `HttpMessageReader`
+exposes a `maxInMemorySize` property and if so the Javadoc will have details about default
+values. In WebFlux, the `ServerCodecConfigurer` provides a
+<<webflux-config-message-codecs,single place>> from where to set all codecs, through the
+`maxInMemorySize` property for default codecs.
+
+For <<webflux-codecs-multipart,Multipart parsing>> the `maxInMemorySize` property limits
+the size of non-file parts. For file parts it determines the threshold at which the part
+is written to disk. For file parts written to disk, there is an additional
+`maxDiskUsagePerPart` property to limit the amount of disk space per part. There is also
+a `maxParts` property to limit the overall number of parts in a multipart request.
+To configure all 3 in WebFlux, you'll need to supply a pre-configured instance of
+`MultipartHttpMessageReader` to `ServerCodecConfigurer`.
+
+
+
 [[webflux-codecs-streaming]]
 ==== Streaming
 [.small]#<<web.adoc#mvc-ann-async-http-streaming, Same as in Spring MVC>>#