Release version 5.1.11.RELEASE

See gh-23884

.gitignore

@@ -24,6 +24,10 @@ buildSrc/build
 /src/asciidoc/build
 target/
 
+# Projects not in this branch
+integration-tests/
+spring-core/kotlin-coroutines/
+
 # Eclipse artifacts, including WTP generated manifests
 .classpath
 .project

build.gradle

@@ -41,11 +41,11 @@ ext {
 	reactorVersion       = "Californium-SR13"
 	rxjavaVersion        = "1.3.8"
 	rxjavaAdapterVersion = "1.2.1"
-	rxjava2Version       = "2.2.12"
+	rxjava2Version       = "2.2.13"
 	slf4jVersion         = "1.7.28"	  // spring-jcl + consistent 3rd party deps
 	tiles3Version        = "3.0.8"
-	tomcatVersion        = "9.0.26"
-	undertowVersion      = "2.0.26.Final"
+	tomcatVersion        = "9.0.27"
+	undertowVersion      = "2.0.27.Final"
 
 	gradleScriptDir = "${rootProject.projectDir}/gradle"
 	withoutJclOverSlf4j = {
@@ -142,7 +142,7 @@ configure(allprojects) { project ->
 	}
 
 	checkstyle {
-		toolVersion = "8.24"
+		toolVersion = "8.26"
 		configDir = rootProject.file("src/checkstyle")
 	}
 

gradle.properties

@@ -1 +1 @@
-version=5.1.11.BUILD-SNAPSHOT
+version=5.1.11.RELEASE

spring-beans/src/main/java/org/springframework/beans/factory/support/BeanDefinitionBuilder.java

@@ -302,6 +302,15 @@ public final class BeanDefinitionBuilder {
 		return this;
 	}
 
+	/**
+	 * Set whether this bean is a primary autowire candidate.
+	 * @since 5.1.11
+	 */
+	public BeanDefinitionBuilder setPrimary(boolean primary) {
+		this.beanDefinition.setPrimary(primary);
+		return this;
+	}
+
 	/**
 	 * Set the role of this definition.
 	 */

spring-context/src/main/java/org/springframework/context/annotation/DeferredImportSelector.java

@@ -16,8 +16,6 @@
 
 package org.springframework.context.annotation;
 
-import java.util.Objects;
-
 import org.springframework.core.type.AnnotationMetadata;
 import org.springframework.lang.Nullable;
 
@@ -100,7 +98,7 @@ public interface DeferredImportSelector extends ImportSelector {
 			}
 
 			@Override
-			public boolean equals(Object other) {
+			public boolean equals(@Nullable Object other) {
 				if (this == other) {
 					return true;
 				}
@@ -108,13 +106,17 @@ public interface DeferredImportSelector extends ImportSelector {
 					return false;
 				}
 				Entry entry = (Entry) other;
-				return (Objects.equals(this.metadata, entry.metadata) &&
-						Objects.equals(this.importClassName, entry.importClassName));
+				return (this.metadata.equals(entry.metadata) && this.importClassName.equals(entry.importClassName));
 			}
 
 			@Override
 			public int hashCode() {
-				return Objects.hash(this.metadata, this.importClassName);
+				return (this.metadata.hashCode() * 31 + this.importClassName.hashCode());
+			}
+
+			@Override
+			public String toString() {
+				return this.importClassName;
 			}
 		}
 	}

spring-context/src/main/java/org/springframework/format/datetime/DateFormatterRegistrar.java

@@ -61,14 +61,13 @@ public class DateFormatterRegistrar implements FormatterRegistrar {
 	@Override
 	public void registerFormatters(FormatterRegistry registry) {
 		addDateConverters(registry);
-		registry.addFormatterForFieldAnnotation(new DateTimeFormatAnnotationFormatterFactory());
-
 		// In order to retain back compatibility we only register Date/Calendar
 		// types when a user defined formatter is specified (see SPR-10105)
 		if (this.dateFormatter != null) {
 			registry.addFormatter(this.dateFormatter);
 			registry.addFormatterForFieldType(Calendar.class, this.dateFormatter);
 		}
+		registry.addFormatterForFieldAnnotation(new DateTimeFormatAnnotationFormatterFactory());
 	}
 
 	/**

spring-context/src/test/java/org/springframework/format/datetime/DateFormattingTests.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.
@@ -46,7 +46,7 @@ import static org.junit.Assert.*;
  */
 public class DateFormattingTests {
 
-	private final FormattingConversionService conversionService = new FormattingConversionService();
+	private FormattingConversionService conversionService;
 
 	private DataBinder binder;
 
@@ -58,6 +58,7 @@ public class DateFormattingTests {
 	}
 
 	private void setup(DateFormatterRegistrar registrar) {
+		conversionService = new FormattingConversionService();
 		DefaultConversionService.addDefaultConverters(conversionService);
 		registrar.registerFormatters(conversionService);
 
@@ -148,6 +149,20 @@ public class DateFormattingTests {
 		assertEquals("10/31/09 1:05", binder.getBindingResult().getFieldValue("dateAnnotatedPattern"));
 	}
 
+	@Test
+	public void testBindDateAnnotatedPatternWithGlobalFormat() {
+		DateFormatterRegistrar registrar = new DateFormatterRegistrar();
+		DateFormatter dateFormatter = new DateFormatter();
+		dateFormatter.setIso(ISO.DATE_TIME);
+		registrar.setFormatter(dateFormatter);
+		setup(registrar);
+		MutablePropertyValues propertyValues = new MutablePropertyValues();
+		propertyValues.add("dateAnnotatedPattern", "10/31/09 1:05");
+		binder.bind(propertyValues);
+		assertEquals(0, binder.getBindingResult().getErrorCount());
+		assertEquals("10/31/09 1:05", binder.getBindingResult().getFieldValue("dateAnnotatedPattern"));
+	}
+
 	@Test
 	public void testBindDateTimeOverflow() {
 		MutablePropertyValues propertyValues = new MutablePropertyValues();

spring-core/src/main/java/org/springframework/core/codec/AbstractDataBufferDecoder.java

@@ -47,12 +47,40 @@ import org.springframework.util.MimeType;
  */
 public abstract class AbstractDataBufferDecoder<T> extends AbstractDecoder<T> {
 
+	private int maxInMemorySize = -1;
+
 
 	protected AbstractDataBufferDecoder(MimeType... supportedMimeTypes) {
 		super(supportedMimeTypes);
 	}
 
 
+	/**
+	 * 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 in 5.1 this is set to -1, unlimited. In 5.2 the default
+	 * value for this limit is set to 256K.
+	 * @param byteCount the max number of bytes to buffer, or -1 for unlimited
+	 * @since 5.1.11
+	 */
+	public void setMaxInMemorySize(int byteCount) {
+		this.maxInMemorySize = byteCount;
+	}
+
+	/**
+	 * Return the {@link #setMaxInMemorySize configured} byte count limit.
+	 * @since 5.1.11
+	 */
+	public int getMaxInMemorySize() {
+		return this.maxInMemorySize;
+	}
+
+
 	@Override
 	public Flux<T> decode(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
@@ -64,7 +92,7 @@ public abstract class AbstractDataBufferDecoder<T> extends AbstractDecoder<T> {
 	public Mono<T> decodeToMono(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
-		return DataBufferUtils.join(input)
+		return DataBufferUtils.join(input, this.maxInMemorySize)
 				.map(buffer -> decodeDataBuffer(buffer, elementType, mimeType, hints));
 	}
 

spring-core/src/main/java/org/springframework/core/codec/StringDecoder.java

@@ -25,14 +25,17 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.function.Consumer;
 
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+import org.springframework.core.io.buffer.LimitedDataBufferList;
 import org.springframework.core.io.buffer.PooledDataBuffer;
 import org.springframework.core.log.LogFormatUtils;
 import org.springframework.lang.Nullable;
@@ -92,10 +95,16 @@ public final class StringDecoder extends AbstractDataBufferDecoder<String> {
 
 		List<byte[]> delimiterBytes = getDelimiterBytes(mimeType);
 
+		// TODO: Drop Consumer and use bufferUntil with Supplier<LimistedDataBufferList> (reactor-core#1925)
+		// TODO: Drop doOnDiscard(LimitedDataBufferList.class, ...) (reactor-core#1924)
+		LimitedDataBufferConsumer limiter = new LimitedDataBufferConsumer(getMaxInMemorySize());
+
 		Flux<DataBuffer> inputFlux = Flux.from(input)
 				.flatMapIterable(buffer -> splitOnDelimiter(buffer, delimiterBytes))
+				.doOnNext(limiter)
 				.bufferUntil(buffer -> buffer == END_FRAME)
 				.map(StringDecoder::joinUntilEndFrame)
+				.doOnDiscard(LimitedDataBufferList.class, LimitedDataBufferList::releaseAndClear)
 				.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
 
 		return super.decode(inputFlux, elementType, mimeType, hints);
@@ -283,4 +292,35 @@ public final class StringDecoder extends AbstractDataBufferDecoder<String> {
 				new MimeType("text", "plain", DEFAULT_CHARSET), MimeTypeUtils.ALL);
 	}
 
+
+	/**
+	 * Temporary measure for reactor-core#1925.
+	 * Consumer that adds to a {@link LimitedDataBufferList} to enforce limits.
+	 */
+	private static class LimitedDataBufferConsumer implements Consumer<DataBuffer> {
+
+		private final LimitedDataBufferList bufferList;
+
+
+		public LimitedDataBufferConsumer(int maxInMemorySize) {
+			this.bufferList = new LimitedDataBufferList(maxInMemorySize);
+		}
+
+
+		@Override
+		public void accept(DataBuffer buffer) {
+			if (buffer == END_FRAME) {
+				this.bufferList.clear();
+			}
+			else {
+				try {
+					this.bufferList.add(buffer);
+				}
+				catch (DataBufferLimitException ex) {
+					DataBufferUtils.release(buffer);
+					throw ex;
+				}
+			}
+		}
+	}
 }

spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferLimitException.java

@@ -0,0 +1,37 @@
+/*
+ * 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.
+ * 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 org.springframework.core.io.buffer;
+
+/**
+ * Exception that indicates the cumulative number of bytes consumed from a
+ * stream of {@link DataBuffer DataBuffer}'s exceeded some pre-configured limit.
+ * This can be raised when data buffers are cached and aggregated, e.g.
+ * {@link DataBufferUtils#join}. Or it could also be raised when data buffers
+ * have been released but a parsed representation is being aggregated, e.g. async
+ * parsing with Jackson.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.1.11
+ */
+@SuppressWarnings("serial")
+public class DataBufferLimitException extends IllegalStateException {
+
+
+	public DataBufferLimitException(String message) {
+		super(message);
+	}
+
+}

spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java

@@ -437,14 +437,36 @@ public abstract class DataBufferUtils {
 	 * @since 5.0.3
 	 */
 	public static Mono<DataBuffer> join(Publisher<DataBuffer> dataBuffers) {
-		Assert.notNull(dataBuffers, "'dataBuffers' must not be null");
+		return join(dataBuffers, -1);
+	}
 
-		return Flux.from(dataBuffers)
-				.collectList()
+	/**
+	 * Variant of {@link #join(Publisher)} that behaves the same way up until
+	 * the specified max number of bytes to buffer. Once the limit is exceeded,
+	 * {@link DataBufferLimitException} is raised.
+	 * @param buffers the data buffers that are to be composed
+	 * @param maxByteCount the max number of bytes to buffer, or -1 for unlimited
+	 * @return a buffer with the aggregated content, possibly an empty Mono if
+	 * the max number of bytes to buffer is exceeded.
+	 * @throws DataBufferLimitException if maxByteCount is exceeded
+	 * @since 5.1.11
+	 */
+	@SuppressWarnings("unchecked")
+	public static Mono<DataBuffer> join(Publisher<? extends DataBuffer> buffers, int maxByteCount) {
+		Assert.notNull(buffers, "'dataBuffers' must not be null");
+
+		if (buffers instanceof Mono) {
+			return (Mono<DataBuffer>) buffers;
+		}
+
+		// TODO: Drop doOnDiscard(LimitedDataBufferList.class, ...) (reactor-core#1924)
+
+		return Flux.from(buffers)
+				.collect(() -> new LimitedDataBufferList(maxByteCount), LimitedDataBufferList::add)
 				.filter(list -> !list.isEmpty())
 				.map(list -> list.get(0).factory().join(list))
+				.doOnDiscard(LimitedDataBufferList.class, LimitedDataBufferList::releaseAndClear)
 				.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
-
 	}
 
 

spring-core/src/main/java/org/springframework/core/io/buffer/LimitedDataBufferList.java

@@ -0,0 +1,157 @@
+/*
+ * 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.
+ * 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 org.springframework.core.io.buffer;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.function.Predicate;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * Custom {@link List} to collect data buffers with and enforce a
+ * limit on the total number of bytes buffered. For use with "collect" or
+ * other buffering operators in declarative APIs, e.g. {@link Flux}.
+ *
+ * <p>Adding elements increases the byte count and if the limit is exceeded,
+ * {@link DataBufferLimitException} is raised.  {@link #clear()} resets the
+ * count. Remove and set are not supported.
+ *
+ * <p><strong>Note:</strong> This class does not automatically release the
+ * buffers it contains. It is usually preferable to use hooks such as
+ * {@link Flux#doOnDiscard} that also take care of cancel and error signals,
+ * or otherwise {@link #releaseAndClear()} can be used.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.1.11
+ */
+@SuppressWarnings("serial")
+public class LimitedDataBufferList extends ArrayList<DataBuffer> {
+
+	private final int maxByteCount;
+
+	private int byteCount;
+
+
+	public LimitedDataBufferList(int maxByteCount) {
+		this.maxByteCount = maxByteCount;
+	}
+
+
+	@Override
+	public boolean add(DataBuffer buffer) {
+		boolean result = super.add(buffer);
+		if (result) {
+			updateCount(buffer.readableByteCount());
+		}
+		return result;
+	}
+
+	@Override
+	public void add(int index, DataBuffer buffer) {
+		super.add(index, buffer);
+		updateCount(buffer.readableByteCount());
+	}
+
+	@Override
+	public boolean addAll(Collection<? extends DataBuffer> collection) {
+		boolean result = super.addAll(collection);
+		collection.forEach(buffer -> updateCount(buffer.readableByteCount()));
+		return result;
+	}
+
+	@Override
+	public boolean addAll(int index, Collection<? extends DataBuffer> collection) {
+		boolean result = super.addAll(index, collection);
+		collection.forEach(buffer -> updateCount(buffer.readableByteCount()));
+		return result;
+	}
+
+	private void updateCount(int bytesToAdd) {
+		if (this.maxByteCount < 0) {
+			return;
+		}
+		if (bytesToAdd > Integer.MAX_VALUE - this.byteCount) {
+			raiseLimitException();
+		}
+		else {
+			this.byteCount += bytesToAdd;
+			if (this.byteCount > this.maxByteCount) {
+				raiseLimitException();
+			}
+		}
+	}
+
+	private void raiseLimitException() {
+		// Do not release here, it's likely down via doOnDiscard..
+		throw new DataBufferLimitException(
+				"Exceeded limit on max bytes to buffer : " + this.maxByteCount);
+	}
+
+	@Override
+	public DataBuffer remove(int index) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public boolean remove(Object o) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	protected void removeRange(int fromIndex, int toIndex) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public boolean removeAll(Collection<?> c) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public boolean removeIf(Predicate<? super DataBuffer> filter) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public DataBuffer set(int index, DataBuffer element) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public void clear() {
+		this.byteCount = 0;
+		super.clear();
+	}
+
+	/**
+	 * Shortcut to {@link DataBufferUtils#release release} all data buffers and
+	 * then {@link #clear()}.
+	 */
+	public void releaseAndClear() {
+		forEach(buf -> {
+			try {
+				DataBufferUtils.release(buf);
+			}
+			catch (Throwable ex) {
+				// Keep going..
+			}
+		});
+		clear();
+	}
+
+}

spring-core/src/main/java/org/springframework/util/unit/DataUnit.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.
@@ -16,38 +16,53 @@
 
 package org.springframework.util.unit;
 
-import java.util.Objects;
-
 /**
- * A standard set of data size units.
+ * A standard set of {@link DataSize} units.
+ *
+ * <p>The unit prefixes used in this class are
+ * <a href="https://en.wikipedia.org/wiki/Binary_prefix">binary prefixes</a>
+ * indicating multiplication by powers of 2. The following table displays the
+ * enum constants defined in this class and corresponding values.
+ *
+ * <p>
+ * <table border="1">
+ * <tr><th>Constant</th><th>Data Size</th><th>Power&nbsp;of&nbsp;2</th><th>Size in Bytes</th></tr>
+ * <tr><td>{@link #BYTES}</td><td>1B</td><td>2^0</td><td>1</td></tr>
+ * <tr><td>{@link #KILOBYTES}</td><td>1KB</td><td>2^10</td><td>1,024</td></tr>
+ * <tr><td>{@link #MEGABYTES}</td><td>1MB</td><td>2^20</td><td>1,048,576</td></tr>
+ * <tr><td>{@link #GIGABYTES}</td><td>1GB</td><td>2^30</td><td>1,073,741,824</td></tr>
+ * <tr><td>{@link #TERABYTES}</td><td>1TB</td><td>2^40</td><td>1,099,511,627,776</td></tr>
+ * </table>
  *
  * @author Stephane Nicoll
+ * @author Sam Brannen
  * @since 5.1
+ * @see DataSize
  */
 public enum DataUnit {
 
 	/**
-	 * Bytes.
+	 * Bytes, represented by suffix {@code B}.
 	 */
 	BYTES("B", DataSize.ofBytes(1)),
 
 	/**
-	 * Kilobytes.
+	 * Kilobytes, represented by suffix {@code KB}.
 	 */
 	KILOBYTES("KB", DataSize.ofKilobytes(1)),
 
 	/**
-	 * Megabytes.
+	 * Megabytes, represented by suffix {@code MB}.
 	 */
 	MEGABYTES("MB", DataSize.ofMegabytes(1)),
 
 	/**
-	 * Gigabytes.
+	 * Gigabytes, represented by suffix {@code GB}.
 	 */
 	GIGABYTES("GB", DataSize.ofGigabytes(1)),
 
 	/**
-	 * Terabytes.
+	 * Terabytes, represented by suffix {@code TB}.
 	 */
 	TERABYTES("TB", DataSize.ofTerabytes(1));
 
@@ -68,18 +83,18 @@ public enum DataUnit {
 
 	/**
 	 * Return the {@link DataUnit} matching the specified {@code suffix}.
-	 * @param suffix one of the standard suffix
+	 * @param suffix one of the standard suffixes
 	 * @return the {@link DataUnit} matching the specified {@code suffix}
-	 * @throws IllegalArgumentException if the suffix does not match any
-	 * of this enum's constants
+	 * @throws IllegalArgumentException if the suffix does not match the suffix
+	 * of any of this enum's constants
 	 */
 	public static DataUnit fromSuffix(String suffix) {
 		for (DataUnit candidate : values()) {
-			if (Objects.equals(candidate.suffix, suffix)) {
+			if (candidate.suffix.equals(suffix)) {
 				return candidate;
 			}
 		}
-		throw new IllegalArgumentException("Unknown unit '" + suffix + "'");
+		throw new IllegalArgumentException("Unknown data unit suffix '" + suffix + "'");
 	}
 
 }

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

@@ -29,6 +29,7 @@ import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
@@ -126,6 +127,20 @@ public class StringDecoderTests extends AbstractDecoderTestCase<StringDecoder> {
 				.verify());
 	}
 
+	@Test
+	public void decodeNewLineWithLimit() {
+		Flux<DataBuffer> input = Flux.just(
+				stringBuffer("abc\n"),
+				stringBuffer("defg\n"),
+				stringBuffer("hijkl\n")
+		);
+		this.decoder.setMaxInMemorySize(4);
+
+		testDecode(input, String.class, step ->
+				step.expectNext("abc", "defg")
+						.verifyError(DataBufferLimitException.class));
+	}
+
 	@Test
 	public void decodeNewLineIncludeDelimiters() {
 		this.decoder = StringDecoder.allMimeTypes(StringDecoder.DEFAULT_DELIMITERS, false);

spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java

@@ -48,11 +48,14 @@ import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.support.DataBufferTestUtils;
 
-import static org.junit.Assert.*;
-import static org.mockito.ArgumentMatchers.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.anyLong;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.isA;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 /**
  * @author Arjen Poutsma
@@ -716,14 +719,25 @@ public class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase {
 		Mono<DataBuffer> result = DataBufferUtils.join(flux);
 
 		StepVerifier.create(result)
-				.consumeNextWith(dataBuffer -> {
-					assertEquals("foobarbaz",
-							DataBufferTestUtils.dumpString(dataBuffer, StandardCharsets.UTF_8));
-					release(dataBuffer);
+				.consumeNextWith(buf -> {
+					assertEquals("foobarbaz", DataBufferTestUtils.dumpString(buf, StandardCharsets.UTF_8));
+					release(buf);
 				})
 				.verifyComplete();
 	}
 
+	@Test
+	public void joinWithLimit() {
+		DataBuffer foo = stringBuffer("foo");
+		DataBuffer bar = stringBuffer("bar");
+		DataBuffer baz = stringBuffer("baz");
+		Flux<DataBuffer> flux = Flux.just(foo, bar, baz);
+		Mono<DataBuffer> result = DataBufferUtils.join(flux, 8);
+
+		StepVerifier.create(result)
+				.verifyError(DataBufferLimitException.class);
+	}
+
 	@Test
 	public void joinErrors() {
 		DataBuffer foo = stringBuffer("foo");

spring-core/src/test/java/org/springframework/core/io/buffer/LimitedDataBufferListTests.java

@@ -0,0 +1,63 @@
+/*
+ * 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.
+ * 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 org.springframework.core.io.buffer;
+
+import java.nio.charset.StandardCharsets;
+
+import org.junit.Test;
+
+import static org.junit.Assert.fail;
+
+/**
+ * Unit tests for {@link LimitedDataBufferList}.
+ * @author Rossen Stoyanchev
+ * @since 5.1.11
+ */
+public class LimitedDataBufferListTests {
+
+	private final static DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
+
+
+	@Test
+	public void limitEnforced() {
+		try {
+			new LimitedDataBufferList(5).add(toDataBuffer("123456"));
+			fail();
+		}
+		catch (DataBufferLimitException ex) {
+			// Expected
+		}
+	}
+
+	@Test
+	public void limitIgnored() {
+		new LimitedDataBufferList(-1).add(toDataBuffer("123456"));
+	}
+
+	@Test
+	public void clearResetsCount() {
+		LimitedDataBufferList list = new LimitedDataBufferList(5);
+		list.add(toDataBuffer("12345"));
+		list.clear();
+		list.add(toDataBuffer("12345"));
+	}
+
+
+	private static DataBuffer toDataBuffer(String value) {
+		return bufferFactory.wrap(value.getBytes(StandardCharsets.UTF_8));
+	}
+
+}

spring-jdbc/src/test/java/org/springframework/jdbc/datasource/DataSourceTransactionManagerTests.java

@@ -858,6 +858,7 @@ public class DataSourceTransactionManagerTests  {
 	public void testTransactionWithIsolationAndReadOnly() throws Exception {
 		given(con.getTransactionIsolation()).willReturn(Connection.TRANSACTION_READ_COMMITTED);
 		given(con.getAutoCommit()).willReturn(true);
+		given(con.isReadOnly()).willReturn(true);
 
 		TransactionTemplate tt = new TransactionTemplate(tm);
 		tt.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
@@ -875,11 +876,13 @@ public class DataSourceTransactionManagerTests  {
 
 		assertTrue("Hasn't thread connection", !TransactionSynchronizationManager.hasResource(ds));
 		InOrder ordered = inOrder(con);
+		ordered.verify(con).setReadOnly(true);
 		ordered.verify(con).setTransactionIsolation(Connection.TRANSACTION_SERIALIZABLE);
 		ordered.verify(con).setAutoCommit(false);
 		ordered.verify(con).commit();
 		ordered.verify(con).setAutoCommit(true);
 		ordered.verify(con).setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
+		ordered.verify(con).setReadOnly(false);
 		verify(con).close();
 	}
 
@@ -890,6 +893,7 @@ public class DataSourceTransactionManagerTests  {
 		given(con.getAutoCommit()).willReturn(true);
 		Statement stmt = mock(Statement.class);
 		given(con.createStatement()).willReturn(stmt);
+		given(con.isReadOnly()).willReturn(true);
 
 		TransactionTemplate tt = new TransactionTemplate(tm);
 		tt.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
@@ -906,11 +910,13 @@ public class DataSourceTransactionManagerTests  {
 
 		assertTrue("Hasn't thread connection", !TransactionSynchronizationManager.hasResource(ds));
 		InOrder ordered = inOrder(con, stmt);
+		ordered.verify(con).setReadOnly(true);
 		ordered.verify(con).setAutoCommit(false);
 		ordered.verify(stmt).executeUpdate("SET TRANSACTION READ ONLY");
 		ordered.verify(stmt).close();
 		ordered.verify(con).commit();
 		ordered.verify(con).setAutoCommit(true);
+		ordered.verify(con).setReadOnly(false);
 		ordered.verify(con).close();
 	}
 

spring-messaging/src/main/java/org/springframework/messaging/simp/user/MultiServerUserRegistry.java

@@ -560,7 +560,6 @@ public class MultiServerUserRegistry implements SimpUserRegistry, SmartApplicati
 			}
 			return map;
 		}
-
 	}
 
 }

spring-orm/spring-orm.gradle

@@ -9,7 +9,7 @@ dependencies {
 	optional(project(":spring-context"))
 	optional(project(":spring-web"))
 	optional("org.eclipse.persistence:org.eclipse.persistence.jpa:2.7.4")
-	optional("org.hibernate:hibernate-core:5.3.12.Final")
+	optional("org.hibernate:hibernate-core:5.3.13.Final")
 	optional("javax.servlet:javax.servlet-api:3.1.0")
 	testCompile("org.aspectj:aspectjweaver:${aspectjVersion}")
 	testCompile("org.hsqldb:hsqldb:${hsqldbVersion}")

spring-test/src/main/java/org/springframework/mock/web/MockCookie.java

@@ -16,6 +16,9 @@
 
 package org.springframework.mock.web;
 
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+
 import javax.servlet.http.Cookie;
 
 import org.springframework.lang.Nullable;
@@ -35,12 +38,15 @@ public class MockCookie extends Cookie {
 	private static final long serialVersionUID = 4312531139502726325L;
 
 
+	@Nullable
+	private ZonedDateTime expires;
+
 	@Nullable
 	private String sameSite;
 
 
 	/**
-	 * Constructor with the cookie name and value.
+	 * Construct a new {@link MockCookie} with the supplied name and value.
 	 * @param name the name
 	 * @param value the value
 	 * @see Cookie#Cookie(String, String)
@@ -49,12 +55,29 @@ public class MockCookie extends Cookie {
 		super(name, value);
 	}
 
+	/**
+	 * Set the "Expires" attribute for this cookie.
+	 * @since 5.1.11
+	 */
+	public void setExpires(@Nullable ZonedDateTime expires) {
+		this.expires = expires;
+	}
 
 	/**
-	 * Add the "SameSite" attribute to the cookie.
+	 * Get the "Expires" attribute for this cookie.
+	 * @since 5.1.11
+	 * @return the "Expires" attribute for this cookie, or {@code null} if not set
+	 */
+	@Nullable
+	public ZonedDateTime getExpires() {
+		return this.expires;
+	}
+
+	/**
+	 * Set the "SameSite" attribute for this cookie.
 	 * <p>This limits the scope of the cookie such that it will only be attached
-	 * to same site requests if {@code "Strict"} or cross-site requests if
-	 * {@code "Lax"}.
+	 * to same-site requests if the supplied value is {@code "Strict"} or cross-site
+	 * requests if the supplied value is {@code "Lax"}.
 	 * @see <a href="https://tools.ietf.org/html/draft-ietf-httpbis-rfc6265bis#section-4.1.2.7">RFC6265 bis</a>
 	 */
 	public void setSameSite(@Nullable String sameSite) {
@@ -62,7 +85,8 @@ public class MockCookie extends Cookie {
 	}
 
 	/**
-	 * Return the "SameSite" attribute, or {@code null} if not set.
+	 * Get the "SameSite" attribute for this cookie.
+	 * @return the "SameSite" attribute for this cookie, or {@code null} if not set
 	 */
 	@Nullable
 	public String getSameSite() {
@@ -71,7 +95,7 @@ public class MockCookie extends Cookie {
 
 
 	/**
-	 * Factory method that parses the value of a "Set-Cookie" header.
+	 * Factory method that parses the value of the supplied "Set-Cookie" header.
 	 * @param setCookieHeader the "Set-Cookie" value; never {@code null} or empty
 	 * @return the created cookie
 	 */
@@ -94,6 +118,10 @@ public class MockCookie extends Cookie {
 			else if (StringUtils.startsWithIgnoreCase(attribute, "Max-Age")) {
 				cookie.setMaxAge(Integer.parseInt(extractAttributeValue(attribute, setCookieHeader)));
 			}
+			else if (StringUtils.startsWithIgnoreCase(attribute, "Expires")) {
+				cookie.setExpires(ZonedDateTime.parse(extractAttributeValue(attribute, setCookieHeader),
+						DateTimeFormatter.RFC_1123_DATE_TIME));
+			}
 			else if (StringUtils.startsWithIgnoreCase(attribute, "Path")) {
 				cookie.setPath(extractAttributeValue(attribute, setCookieHeader));
 			}

spring-test/src/main/java/org/springframework/mock/web/MockHttpServletResponse.java

@@ -26,6 +26,8 @@ import java.io.Writer;
 import java.text.DateFormat;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -345,9 +347,15 @@ public class MockHttpServletResponse implements HttpServletResponse {
 		if (maxAge >= 0) {
 			buf.append("; Max-Age=").append(maxAge);
 			buf.append("; Expires=");
-			HttpHeaders headers = new HttpHeaders();
-			headers.setExpires(maxAge > 0 ? System.currentTimeMillis() + 1000L * maxAge : 0);
-			buf.append(headers.getFirst(HttpHeaders.EXPIRES));
+			ZonedDateTime expires = (cookie instanceof MockCookie ? ((MockCookie) cookie).getExpires() : null);
+			if (expires != null) {
+				buf.append(expires.format(DateTimeFormatter.RFC_1123_DATE_TIME));
+			}
+			else {
+				HttpHeaders headers = new HttpHeaders();
+				headers.setExpires(maxAge > 0 ? System.currentTimeMillis() + 1000L * maxAge : 0);
+				buf.append(headers.getFirst(HttpHeaders.EXPIRES));
+			}
 		}
 
 		if (cookie.getSecure()) {

spring-test/src/main/java/org/springframework/mock/web/MockServletContext.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.
@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.nio.file.InvalidPathException;
 import java.util.Collections;
 import java.util.Enumeration;
 import java.util.EventListener;
@@ -294,8 +295,10 @@ public class MockServletContext implements ServletContext {
 	@Nullable
 	public Set<String> getResourcePaths(String path) {
 		String actualPath = (path.endsWith("/") ? path : path + "/");
-		Resource resource = this.resourceLoader.getResource(getResourceLocation(actualPath));
+		String resourceLocation = getResourceLocation(actualPath);
+		Resource resource = null;
 		try {
+			resource = this.resourceLoader.getResource(resourceLocation);
 			File file = resource.getFile();
 			String[] fileList = file.list();
 			if (ObjectUtils.isEmpty(fileList)) {
@@ -311,9 +314,10 @@ public class MockServletContext implements ServletContext {
 			}
 			return resourcePaths;
 		}
-		catch (IOException ex) {
+		catch (InvalidPathException | IOException ex ) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("Could not get resource paths for " + resource, ex);
+				logger.warn("Could not get resource paths for " +
+						(resource != null ? resource : resourceLocation), ex);
 			}
 			return null;
 		}
@@ -322,19 +326,22 @@ public class MockServletContext implements ServletContext {
 	@Override
 	@Nullable
 	public URL getResource(String path) throws MalformedURLException {
-		Resource resource = this.resourceLoader.getResource(getResourceLocation(path));
-		if (!resource.exists()) {
-			return null;
-		}
+		String resourceLocation = getResourceLocation(path);
+		Resource resource = null;
 		try {
+			resource = this.resourceLoader.getResource(resourceLocation);
+			if (!resource.exists()) {
+				return null;
+			}
 			return resource.getURL();
 		}
 		catch (MalformedURLException ex) {
 			throw ex;
 		}
-		catch (IOException ex) {
+		catch (InvalidPathException | IOException ex) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("Could not get URL for " + resource, ex);
+				logger.warn("Could not get URL for resource " +
+						(resource != null ? resource : resourceLocation), ex);
 			}
 			return null;
 		}
@@ -343,16 +350,19 @@ public class MockServletContext implements ServletContext {
 	@Override
 	@Nullable
 	public InputStream getResourceAsStream(String path) {
-		Resource resource = this.resourceLoader.getResource(getResourceLocation(path));
-		if (!resource.exists()) {
-			return null;
-		}
+		String resourceLocation = getResourceLocation(path);
+		Resource resource = null;
 		try {
+			resource = this.resourceLoader.getResource(resourceLocation);
+			if (!resource.exists()) {
+				return null;
+			}
 			return resource.getInputStream();
 		}
-		catch (IOException ex) {
+		catch (InvalidPathException | IOException ex) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("Could not open InputStream for " + resource, ex);
+				logger.warn("Could not open InputStream for resource " +
+						(resource != null ? resource : resourceLocation), ex);
 			}
 			return null;
 		}
@@ -459,13 +469,16 @@ public class MockServletContext implements ServletContext {
 	@Override
 	@Nullable
 	public String getRealPath(String path) {
-		Resource resource = this.resourceLoader.getResource(getResourceLocation(path));
+		String resourceLocation = getResourceLocation(path);
+		Resource resource = null;
 		try {
+			resource = this.resourceLoader.getResource(resourceLocation);
 			return resource.getFile().getAbsolutePath();
 		}
-		catch (IOException ex) {
+		catch (InvalidPathException | IOException ex) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("Could not determine real path of resource " + resource, ex);
+				logger.warn("Could not determine real path of resource " +
+						(resource != null ? resource : resourceLocation), ex);
 			}
 			return null;
 		}

spring-test/src/test/java/org/springframework/mock/web/MockCookieTests.java

@@ -20,6 +20,9 @@ import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
 
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+
 import static org.junit.Assert.*;
 
 /**
@@ -67,8 +70,8 @@ public class MockCookieTests {
 
 	@Test
 	public void parseHeaderWithAttributes() {
-		MockCookie cookie = MockCookie.parse(
-				"SESSION=123; Domain=example.com; Max-Age=60; Path=/; Secure; HttpOnly; SameSite=Lax");
+		MockCookie cookie = MockCookie.parse("SESSION=123; Domain=example.com; Max-Age=60; " +
+				"Expires=Tue, 8 Oct 2019 19:50:00 GMT; Path=/; Secure; HttpOnly; SameSite=Lax");
 
 		assertCookie(cookie, "SESSION", "123");
 		assertEquals("example.com", cookie.getDomain());
@@ -76,6 +79,8 @@ public class MockCookieTests {
 		assertEquals("/", cookie.getPath());
 		assertTrue(cookie.getSecure());
 		assertTrue(cookie.isHttpOnly());
+		assertEquals(ZonedDateTime.parse("Tue, 8 Oct 2019 19:50:00 GMT",
+				DateTimeFormatter.RFC_1123_DATE_TIME), cookie.getExpires());
 		assertEquals("Lax", cookie.getSameSite());
 	}
 
@@ -109,15 +114,17 @@ public class MockCookieTests {
 
 	@Test
 	public void parseHeaderWithAttributesCaseSensitivity() {
-		MockCookie cookie = MockCookie.parse(
-				"SESSION=123; domain=example.com; max-age=60; path=/; secure; httponly; samesite=Lax");
-
+		MockCookie cookie = MockCookie.parse("SESSION=123; domain=example.com; max-age=60; " +
+				"expires=Tue, 8 Oct 2019 19:50:00 GMT; path=/; secure; httponly; samesite=Lax");
+		
 		assertCookie(cookie, "SESSION", "123");
 		assertEquals("example.com", cookie.getDomain());
 		assertEquals(60, cookie.getMaxAge());
 		assertEquals("/", cookie.getPath());
 		assertTrue(cookie.getSecure());
 		assertTrue(cookie.isHttpOnly());
+		assertEquals(ZonedDateTime.parse("Tue, 8 Oct 2019 19:50:00 GMT",
+				DateTimeFormatter.RFC_1123_DATE_TIME), cookie.getExpires());
 		assertEquals("Lax", cookie.getSameSite());
 	}
 

spring-test/src/test/java/org/springframework/mock/web/MockHttpServletResponseTests.java

@@ -39,6 +39,7 @@ import static org.junit.Assert.*;
  * @author Rob Winch
  * @author Sam Brannen
  * @author Brian Clozel
+ * @author Vedran Pavic
  * @since 19.02.2006
  */
 public class MockHttpServletResponseTests {
@@ -338,6 +339,18 @@ public class MockHttpServletResponseTests {
 		assertPrimarySessionCookie("999");
 	}
 
+	/**
+	 * @since 5.1.11
+	 */
+	@Test
+	public void setCookieHeaderWithExpiresAttribute() {
+		String cookieValue = "SESSION=123; Path=/; Max-Age=100; Expires=Tue, 8 Oct 2019 19:50:00 GMT; Secure; " +
+				"HttpOnly; SameSite=Lax";
+		response.setHeader(HttpHeaders.SET_COOKIE, cookieValue);
+		assertNumCookies(1);
+		assertEquals(cookieValue, response.getHeader(HttpHeaders.SET_COOKIE));
+	}
+
 	@Test
 	public void addCookieHeader() {
 		response.addHeader(HttpHeaders.SET_COOKIE, "SESSION=123; Path=/; Secure; HttpOnly; SameSite=Lax");
@@ -351,6 +364,17 @@ public class MockHttpServletResponseTests {
 		assertCookieValues("123", "999");
 	}
 
+	/**
+	 * @since 5.1.11
+	 */
+	@Test
+	public void addCookieHeaderWithExpiresAttribute() {
+		String cookieValue = "SESSION=123; Path=/; Max-Age=100; Expires=Tue, 8 Oct 2019 19:50:00 GMT; Secure; " +
+				"HttpOnly; SameSite=Lax";
+		response.addHeader(HttpHeaders.SET_COOKIE, cookieValue);
+		assertEquals(cookieValue, response.getHeader(HttpHeaders.SET_COOKIE));
+	}
+
 	@Test
 	public void addCookie() {
 		MockCookie mockCookie = new MockCookie("SESSION", "123");

spring-test/src/test/java/org/springframework/mock/web/MockServletContextTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.mock.web;
 
+import java.io.InputStream;
+import java.net.URL;
 import java.util.Map;
 import java.util.Set;
 
@@ -24,7 +26,9 @@ import javax.servlet.RequestDispatcher;
 import javax.servlet.ServletRegistration;
 
 import org.junit.Test;
+import org.junit.jupiter.api.condition.OS;
 
+import org.springframework.core.io.FileSystemResourceLoader;
 import org.springframework.http.MediaType;
 
 import static org.junit.Assert.assertEquals;
@@ -34,6 +38,8 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
 /**
+ * Unit tests for {@link MockServletContext}.
+ *
  * @author Juergen Hoeller
  * @author Chris Beams
  * @author Sam Brannen
@@ -45,27 +51,27 @@ public class MockServletContextTests {
 
 
 	@Test
-	public void listFiles() {
+	public void getResourcePaths() {
 		Set<String> paths = sc.getResourcePaths("/web");
 		assertNotNull(paths);
 		assertTrue(paths.contains("/web/MockServletContextTests.class"));
 	}
 
 	@Test
-	public void listSubdirectories() {
+	public void getResourcePathsWithSubdirectories() {
 		Set<String> paths = sc.getResourcePaths("/");
 		assertNotNull(paths);
 		assertTrue(paths.contains("/web/"));
 	}
 
 	@Test
-	public void listNonDirectory() {
+	public void getResourcePathsWithNonDirectory() {
 		Set<String> paths = sc.getResourcePaths("/web/MockServletContextTests.class");
 		assertNull(paths);
 	}
 
 	@Test
-	public void listInvalidPath() {
+	public void getResourcePathsWithInvalidPath() {
 		Set<String> paths = sc.getResourcePaths("/web/invalid");
 		assertNull(paths);
 	}
@@ -194,4 +200,50 @@ public class MockServletContextTests {
 		assertEquals(0, filterRegistrations.size());
 	}
 
+	/**
+	 * @since 5.1.11
+	 */
+	@Test
+	public void getResourcePathsWithRelativePathToWindowsCDrive() {
+		MockServletContext servletContext = new MockServletContext( "org/springframework/mock", new FileSystemResourceLoader());
+		Set<String> paths = servletContext.getResourcePaths("C:\\temp");
+		assertNull(paths);
+	}
+
+	/**
+	 * @since 5.1.11
+	 */
+	@Test
+	public void getResourceWithRelativePathToWindowsCDrive() throws Exception {
+		MockServletContext servletContext = new MockServletContext( "org/springframework/mock", new FileSystemResourceLoader());
+		URL resource = servletContext.getResource("C:\\temp");
+		assertNull(resource);
+	}
+
+	/**
+	 * @since 5.1.11
+	 */
+	@Test
+	public void getResourceAsStreamWithRelativePathToWindowsCDrive() {
+		MockServletContext servletContext = new MockServletContext( "org/springframework/mock", new FileSystemResourceLoader());
+		InputStream inputStream = servletContext.getResourceAsStream("C:\\temp");
+		assertNull(inputStream);
+	}
+
+	/**
+	 * @since 5.1.11
+	 */
+	@Test
+	public void getRealPathWithRelativePathToWindowsCDrive() {
+		MockServletContext servletContext = new MockServletContext( "org/springframework/mock", new FileSystemResourceLoader());
+		String realPath = servletContext.getRealPath("C:\\temp");
+
+		if (OS.WINDOWS.isCurrentOs()) {
+			assertNull(realPath);
+		}
+		else {
+			assertNotNull(realPath);
+		}
+	}
+
 }

spring-tx/src/test/java/org/springframework/transaction/interceptor/TransactionInterceptorTests.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.
@@ -40,6 +40,7 @@ import static org.mockito.BDDMockito.*;
  * Mock object based tests for TransactionInterceptor.
  *
  * @author Rod Johnson
+ * @author Juergen Hoeller
  * @since 16.03.2003
  */
 public class TransactionInterceptorTests extends AbstractTransactionAspectTests {
@@ -49,7 +50,7 @@ public class TransactionInterceptorTests extends AbstractTransactionAspectTests
 
 
 	@Override
-	protected Object advised(Object target, PlatformTransactionManager ptm, TransactionAttributeSource[] tas) throws Exception {
+	protected Object advised(Object target, PlatformTransactionManager ptm, TransactionAttributeSource[] tas) {
 		TransactionInterceptor ti = new TransactionInterceptor();
 		ti.setTransactionManager(ptm);
 		ti.setTransactionAttributeSources(tas);

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,22 @@ 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. In 5.1 most codecs are not limited except {@code FormHttpMessageReader}
+		 * which is limited to 256K. In 5.2 all codecs are limited to 256K by default.
+		 * @param byteCount the max number of bytes to buffer, or -1 for unlimited
+		 * @since 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/FormHttpMessageReader.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.
@@ -30,6 +30,7 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.Hints;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.log.LogFormatUtils;
 import org.springframework.http.MediaType;
@@ -62,6 +63,8 @@ public class FormHttpMessageReader extends LoggingCodecSupport
 
 	private Charset defaultCharset = DEFAULT_CHARSET;
 
+	private int maxInMemorySize = 256 * 1024;
+
 
 	/**
 	 * Set the default character set to use for reading form data when the
@@ -80,6 +83,26 @@ public class FormHttpMessageReader extends LoggingCodecSupport
 		return this.defaultCharset;
 	}
 
+	/**
+	 * Set the max number of bytes for input form data. As form data is buffered
+	 * before it is parsed, this helps to limit the amount of buffering. Once
+	 * the limit is exceeded, {@link DataBufferLimitException} is raised.
+	 * <p>By default this is set to 256K.
+	 * @param byteCount the max number of bytes to buffer, or -1 for unlimited
+	 * @since 5.1.11
+	 */
+	public void setMaxInMemorySize(int byteCount) {
+		this.maxInMemorySize = byteCount;
+	}
+
+	/**
+	 * Return the {@link #setMaxInMemorySize configured} byte count limit.
+	 * @since 5.1.11
+	 */
+	public int getMaxInMemorySize() {
+		return this.maxInMemorySize;
+	}
+
 
 	@Override
 	public boolean canRead(ResolvableType elementType, @Nullable MediaType mediaType) {
@@ -105,7 +128,7 @@ public class FormHttpMessageReader extends LoggingCodecSupport
 		MediaType contentType = message.getHeaders().getContentType();
 		Charset charset = getMediaTypeCharset(contentType);
 
-		return DataBufferUtils.join(message.getBody())
+		return DataBufferUtils.join(message.getBody(), getMaxInMemorySize())
 				.map(buffer -> {
 					CharBuffer charBuffer = charset.decode(buffer.asByteBuffer());
 					String body = charBuffer.toString();

spring-web/src/main/java/org/springframework/http/codec/ServerCodecConfigurer.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.
@@ -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/json/AbstractJackson2Decoder.java

@@ -38,6 +38,7 @@ import org.springframework.core.codec.CodecException;
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.codec.Hints;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.log.LogFormatUtils;
 import org.springframework.http.codec.HttpMessageDecoder;
 import org.springframework.http.server.reactive.ServerHttpRequest;
@@ -57,6 +58,9 @@ import org.springframework.util.MimeType;
  */
 public abstract class AbstractJackson2Decoder extends Jackson2CodecSupport implements HttpMessageDecoder<Object> {
 
+	private int maxInMemorySize = -1;
+
+
 	/**
 	 * Until https://github.com/FasterXML/jackson-core/issues/476 is resolved,
 	 * we need to ensure buffer recycling is off.
@@ -74,6 +78,29 @@ public abstract class AbstractJackson2Decoder extends Jackson2CodecSupport imple
 	}
 
 
+	/**
+	 * Set the max number of bytes that can be buffered by this decoder. This
+	 * is either the size of the entire input when decoding as a whole, or the
+	 * size of one top-level JSON object within a JSON stream. When the limit
+	 * is exceeded, {@link DataBufferLimitException} is raised.
+	 * <p>By default in 5.1 this is set to -1, unlimited. In 5.2 the default
+	 * value for this limit is set to 256K.
+	 * @param byteCount the max number of bytes to buffer, or -1 for unlimited
+	 * @since 5.1.11
+	 */
+	public void setMaxInMemorySize(int byteCount) {
+		this.maxInMemorySize = byteCount;
+	}
+
+	/**
+	 * Return the {@link #setMaxInMemorySize configured} byte count limit.
+	 * @since 5.1.11
+	 */
+	public int getMaxInMemorySize() {
+		return this.maxInMemorySize;
+	}
+
+
 	@Override
 	public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType) {
 		JavaType javaType = getObjectMapper().getTypeFactory().constructType(elementType.getType());
@@ -87,7 +114,7 @@ public abstract class AbstractJackson2Decoder extends Jackson2CodecSupport imple
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
 		Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(
-				Flux.from(input), this.jsonFactory, getObjectMapper(), true);
+				Flux.from(input), this.jsonFactory, getObjectMapper(), true, getMaxInMemorySize());
 		return decodeInternal(tokens, elementType, mimeType, hints);
 	}
 
@@ -96,7 +123,7 @@ public abstract class AbstractJackson2Decoder extends Jackson2CodecSupport imple
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
 		Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(
-				Flux.from(input), this.jsonFactory, getObjectMapper(), false);
+				Flux.from(input), this.jsonFactory, getObjectMapper(), false, getMaxInMemorySize());
 		return decodeInternal(tokens, elementType, mimeType, hints).singleOrEmpty();
 	}
 

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

@@ -34,6 +34,7 @@ import reactor.core.publisher.Flux;
 
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.io.buffer.DataBufferUtils;
 
 /**
@@ -60,30 +61,39 @@ final class Jackson2Tokenizer {
 
 	private int arrayDepth;
 
+	private final int maxInMemorySize;
+
+	private int byteCount;
+
+
 	// TODO: change to ByteBufferFeeder when supported by Jackson
 	// See https://github.com/FasterXML/jackson-core/issues/478
 	private final ByteArrayFeeder inputFeeder;
 
 
-	private Jackson2Tokenizer(
-			JsonParser parser, DeserializationContext deserializationContext, boolean tokenizeArrayElements) {
+	private Jackson2Tokenizer(JsonParser parser, DeserializationContext deserializationContext,
+			boolean tokenizeArrayElements, int maxInMemorySize) {
 
 		this.parser = parser;
 		this.deserializationContext = deserializationContext;
 		this.tokenizeArrayElements = tokenizeArrayElements;
 		this.tokenBuffer = new TokenBuffer(parser, deserializationContext);
 		this.inputFeeder = (ByteArrayFeeder) this.parser.getNonBlockingInputFeeder();
+		this.maxInMemorySize = maxInMemorySize;
 	}
 
 
 	private Flux<TokenBuffer> tokenize(DataBuffer dataBuffer) {
+		int bufferSize = dataBuffer.readableByteCount();
 		byte[] bytes = new byte[dataBuffer.readableByteCount()];
 		dataBuffer.read(bytes);
 		DataBufferUtils.release(dataBuffer);
 
 		try {
 			this.inputFeeder.feedInput(bytes, 0, bytes.length);
-			return parseTokenBufferFlux();
+			List<TokenBuffer> result = parseTokenBufferFlux();
+			assertInMemorySize(bufferSize, result);
+			return Flux.fromIterable(result);
 		}
 		catch (JsonProcessingException ex) {
 			return Flux.error(new DecodingException("JSON decoding error: " + ex.getOriginalMessage(), ex));
@@ -96,7 +106,8 @@ final class Jackson2Tokenizer {
 	private Flux<TokenBuffer> endOfInput() {
 		this.inputFeeder.endOfInput();
 		try {
-			return parseTokenBufferFlux();
+			List<TokenBuffer> result = parseTokenBufferFlux();
+			return Flux.fromIterable(result);
 		}
 		catch (JsonProcessingException ex) {
 			return Flux.error(new DecodingException("JSON decoding error: " + ex.getOriginalMessage(), ex));
@@ -106,7 +117,7 @@ final class Jackson2Tokenizer {
 		}
 	}
 
-	private Flux<TokenBuffer> parseTokenBufferFlux() throws IOException {
+	private List<TokenBuffer> parseTokenBufferFlux() throws IOException {
 		List<TokenBuffer> result = new ArrayList<>();
 
 		while (true) {
@@ -124,7 +135,7 @@ final class Jackson2Tokenizer {
 				processTokenArray(token, result);
 			}
 		}
-		return Flux.fromIterable(result);
+		return result;
 	}
 
 	private void updateDepth(JsonToken token) {
@@ -171,18 +182,40 @@ final class Jackson2Tokenizer {
 				(token == JsonToken.END_ARRAY && this.arrayDepth == 0));
 	}
 
+	private void assertInMemorySize(int currentBufferSize, List<TokenBuffer> result) {
+		if (this.maxInMemorySize >= 0) {
+			if (!result.isEmpty()) {
+				this.byteCount = 0;
+			}
+			else if (currentBufferSize > Integer.MAX_VALUE - this.byteCount) {
+				raiseLimitException();
+			}
+			else {
+				this.byteCount += currentBufferSize;
+				if (this.byteCount > this.maxInMemorySize) {
+					raiseLimitException();
+				}
+			}
+		}
+	}
+
+	private void raiseLimitException() {
+		throw new DataBufferLimitException(
+				"Exceeded limit on max bytes per JSON object: " + this.maxInMemorySize);
+	}
+
 
 	/**
 	 * Tokenize the given {@code Flux<DataBuffer>} into {@code Flux<TokenBuffer>}.
 	 * @param dataBuffers the source data buffers
 	 * @param jsonFactory the factory to use
 	 * @param objectMapper the current mapper instance
-	 * @param tokenizeArrayElements if {@code true} and the "top level" JSON object is
+	 * @param tokenizeArrays if {@code true} and the "top level" JSON object is
 	 * an array, each element is returned individually immediately after it is received
 	 * @return the resulting token buffers
 	 */
 	public static Flux<TokenBuffer> tokenize(Flux<DataBuffer> dataBuffers, JsonFactory jsonFactory,
-			ObjectMapper objectMapper, boolean tokenizeArrayElements) {
+			ObjectMapper objectMapper, boolean tokenizeArrays, int maxInMemorySize) {
 
 		try {
 			JsonParser parser = jsonFactory.createNonBlockingByteArrayParser();
@@ -191,7 +224,7 @@ final class Jackson2Tokenizer {
 				context = ((DefaultDeserializationContext) context).createInstance(
 						objectMapper.getDeserializationConfig(), parser, objectMapper.getInjectableValues());
 			}
-			Jackson2Tokenizer tokenizer = new Jackson2Tokenizer(parser, context, tokenizeArrayElements);
+			Jackson2Tokenizer tokenizer = new Jackson2Tokenizer(parser, context, tokenizeArrays, maxInMemorySize);
 			return dataBuffers.flatMap(tokenizer::tokenize, Flux::error, tokenizer::endOfInput);
 		}
 		catch (IOException ex) {

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/multipart/SynchronossPartHttpMessageReader.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.
@@ -40,14 +40,18 @@ import org.synchronoss.cloud.nio.multipart.NioMultipartParser;
 import org.synchronoss.cloud.nio.multipart.NioMultipartParserListener;
 import org.synchronoss.cloud.nio.multipart.PartBodyStreamStorageFactory;
 import org.synchronoss.cloud.nio.stream.storage.StreamStorage;
+import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.FluxSink;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.SignalType;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.DecodingException;
 import org.springframework.core.codec.Hints;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.core.log.LogFormatUtils;
@@ -69,15 +73,83 @@ import org.springframework.util.Assert;
  * @author Sebastien Deleuze
  * @author Rossen Stoyanchev
  * @author Arjen Poutsma
+ * @author Brian Clozel
  * @since 5.0
  * @see <a href="https://github.com/synchronoss/nio-multipart">Synchronoss NIO Multipart</a>
  * @see MultipartHttpMessageReader
  */
 public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implements HttpMessageReader<Part> {
 
-	private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
+	// Static DataBufferFactory to copy from FileInputStream or wrap bytes[].
+	private static final DataBufferFactory bufferFactory = new DefaultDataBufferFactory();
 
-	private final PartBodyStreamStorageFactory streamStorageFactory = new DefaultPartBodyStreamStorageFactory();
+
+	private int maxInMemorySize = -1;
+
+	private long maxDiskUsagePerPart = -1;
+
+	private int maxParts = -1;
+
+
+	/**
+	 * Configure the maximum amount of memory that is allowed to use per part.
+	 * When the limit is exceeded:
+	 * <ul>
+	 * <li>file parts are written to a temporary file.
+	 * <li>non-file parts are rejected with {@link DataBufferLimitException}.
+	 * </ul>
+	 * <p>By default in 5.1 this is set to -1 in which case this limit is
+	 * not enforced and all parts may be written to disk and are limited only
+	 * by the {@link #setMaxDiskUsagePerPart(long) maxDiskUsagePerPart} property.
+	 * In 5.2 this default value for this limit is set to 256K.
+	 * @param byteCount the in-memory limit in bytes, or -1 for unlimited
+	 * @since 5.1.11
+	 */
+	public void setMaxInMemorySize(int byteCount) {
+		this.maxInMemorySize = byteCount;
+	}
+
+	/**
+	 * Get the {@link #setMaxInMemorySize configured} maximum in-memory size.
+	 * @since 5.1.11
+	 */
+	public int getMaxInMemorySize() {
+		return this.maxInMemorySize;
+	}
+
+	/**
+	 * Configure the maximum amount of disk space allowed for file parts.
+	 * <p>By default this is set to -1.
+	 * @param maxDiskUsagePerPart the disk limit in bytes, or -1 for unlimited
+	 * @since 5.1.11
+	 */
+	public void setMaxDiskUsagePerPart(long maxDiskUsagePerPart) {
+		this.maxDiskUsagePerPart = maxDiskUsagePerPart;
+	}
+
+	/**
+	 * Get the {@link #setMaxDiskUsagePerPart configured} maximum disk usage.
+	 * @since 5.1.11
+	 */
+	public long getMaxDiskUsagePerPart() {
+		return this.maxDiskUsagePerPart;
+	}
+
+	/**
+	 * Specify the maximum number of parts allowed in a given multipart request.
+	 * @since 5.1.11
+	 */
+	public void setMaxParts(int maxParts) {
+		this.maxParts = maxParts;
+	}
+
+	/**
+	 * Return the {@link #setMaxParts configured} limit on the number of parts.
+	 * @since 5.1.11
+	 */
+	public int getMaxParts() {
+		return this.maxParts;
+	}
 
 
 	@Override
@@ -91,10 +163,9 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 				(mediaType == null || MediaType.MULTIPART_FORM_DATA.isCompatibleWith(mediaType));
 	}
 
-
 	@Override
 	public Flux<Part> read(ResolvableType elementType, ReactiveHttpInputMessage message, Map<String, Object> hints) {
-		return Flux.create(new SynchronossPartGenerator(message, this.bufferFactory, this.streamStorageFactory))
+		return Flux.create(new SynchronossPartGenerator(message))
 				.doOnNext(part -> {
 					if (!Hints.isLoggingSuppressed(hints)) {
 						LogFormatUtils.traceDebug(logger, traceOn -> Hints.getLogPrefix(hints) + "Parsed " +
@@ -105,7 +176,6 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 				});
 	}
 
-
 	@Override
 	public Mono<Part> readMono(ResolvableType elementType, ReactiveHttpInputMessage message, Map<String, Object> hints) {
 		return Mono.error(new UnsupportedOperationException("Cannot read multipart request body into single Part"));
@@ -113,27 +183,27 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 
 
 	/**
-	 * Consume and feed input to the Synchronoss parser, then listen for parser
-	 * output events and adapt to {@code Flux<Sink<Part>>}.
+	 * Subscribe to the input stream and feed the Synchronoss parser. Then listen
+	 * for parser output, creating parts, and pushing them into the FluxSink.
 	 */
-	private static class SynchronossPartGenerator implements Consumer<FluxSink<Part>> {
+	private class SynchronossPartGenerator extends BaseSubscriber<DataBuffer> implements Consumer<FluxSink<Part>> {
 
 		private final ReactiveHttpInputMessage inputMessage;
 
-		private final DataBufferFactory bufferFactory;
+		private final LimitedPartBodyStreamStorageFactory storageFactory = new LimitedPartBodyStreamStorageFactory();
 
-		private final PartBodyStreamStorageFactory streamStorageFactory;
+		@Nullable
+		private NioMultipartParserListener listener;
 
-		SynchronossPartGenerator(ReactiveHttpInputMessage inputMessage, DataBufferFactory bufferFactory,
-				PartBodyStreamStorageFactory streamStorageFactory) {
+		@Nullable
+		private NioMultipartParser parser;
 
+		public SynchronossPartGenerator(ReactiveHttpInputMessage inputMessage) {
 			this.inputMessage = inputMessage;
-			this.bufferFactory = bufferFactory;
-			this.streamStorageFactory = streamStorageFactory;
 		}
 
 		@Override
-		public void accept(FluxSink<Part> emitter) {
+		public void accept(FluxSink<Part> sink) {
 			HttpHeaders headers = this.inputMessage.getHeaders();
 			MediaType mediaType = headers.getContentType();
 			Assert.state(mediaType != null, "No content type set");
@@ -142,40 +212,68 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 			Charset charset = Optional.ofNullable(mediaType.getCharset()).orElse(StandardCharsets.UTF_8);
 			MultipartContext context = new MultipartContext(mediaType.toString(), length, charset.name());
 
-			NioMultipartParserListener listener = new FluxSinkAdapterListener(emitter, this.bufferFactory, context);
-			NioMultipartParser parser = Multipart
-					.multipart(context)
-					.usePartBodyStreamStorageFactory(this.streamStorageFactory)
-					.forNIO(listener);
+			this.listener = new FluxSinkAdapterListener(sink, context, this.storageFactory);
 
-			this.inputMessage.getBody().subscribe(buffer -> {
-				byte[] resultBytes = new byte[buffer.readableByteCount()];
-				buffer.read(resultBytes);
-				try {
-					parser.write(resultBytes);
+			this.parser = Multipart
+					.multipart(context)
+					.usePartBodyStreamStorageFactory(this.storageFactory)
+					.forNIO(this.listener);
+
+			this.inputMessage.getBody().subscribe(this);
+		}
+
+		@Override
+		protected void hookOnNext(DataBuffer buffer) {
+			Assert.state(this.parser != null && this.listener != null, "Not initialized yet");
+
+			int size = buffer.readableByteCount();
+			this.storageFactory.increaseByteCount(size);
+			byte[] resultBytes = new byte[size];
+			buffer.read(resultBytes);
+
+			try {
+				this.parser.write(resultBytes);
+			}
+			catch (IOException ex) {
+				cancel();
+				int index = this.storageFactory.getCurrentPartIndex();
+				this.listener.onError("Parser error for part [" + index + "]", ex);
+			}
+			finally {
+				DataBufferUtils.release(buffer);
+			}
+		}
+
+		@Override
+		protected void hookOnError(Throwable ex) {
+			try {
+				if (this.parser != null) {
+					this.parser.close();
 				}
-				catch (IOException ex) {
-					listener.onError("Exception thrown providing input to the parser", ex);
+			}
+			catch (IOException ex2) {
+				// ignore
+			}
+			finally {
+				if (this.listener != null) {
+					int index = this.storageFactory.getCurrentPartIndex();
+					this.listener.onError("Failure while parsing part[" + index + "]", ex);
 				}
-				finally {
-					DataBufferUtils.release(buffer);
+			}
+		}
+
+		@Override
+		protected void hookFinally(SignalType type) {
+			try {
+				if (this.parser != null) {
+					this.parser.close();
 				}
-			}, ex -> {
-				try {
-					listener.onError("Request body input error", ex);
-					parser.close();
+			}
+			catch (IOException ex) {
+				if (this.listener != null) {
+					this.listener.onError("Error while closing parser", ex);
 				}
-				catch (IOException ex2) {
-					listener.onError("Exception thrown while closing the parser", ex2);
-				}
-			}, () -> {
-				try {
-					parser.close();
-				}
-				catch (IOException ex) {
-					listener.onError("Exception thrown while closing the parser", ex);
-				}
-			});
+			}
 		}
 
 		private int getContentLength(HttpHeaders headers) {
@@ -186,6 +284,53 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 	}
 
 
+	private class LimitedPartBodyStreamStorageFactory implements PartBodyStreamStorageFactory {
+
+		private final PartBodyStreamStorageFactory storageFactory = (maxInMemorySize > 0 ?
+				new DefaultPartBodyStreamStorageFactory(maxInMemorySize) :
+				new DefaultPartBodyStreamStorageFactory());
+
+		private int index = 1;
+
+		private boolean isFilePart;
+
+		private long partSize;
+
+		public int getCurrentPartIndex() {
+			return this.index;
+		}
+
+		@Override
+		public StreamStorage newStreamStorageForPartBody(Map<String, List<String>> headers, int index) {
+			this.index = index;
+			this.isFilePart = (MultipartUtils.getFileName(headers) != null);
+			this.partSize = 0;
+			if (maxParts > 0 && index > maxParts) {
+				throw new DecodingException("Too many parts (" + index + " allowed)");
+			}
+			return this.storageFactory.newStreamStorageForPartBody(headers, index);
+		}
+
+		public void increaseByteCount(long byteCount) {
+			this.partSize += byteCount;
+			if (maxInMemorySize > 0 && !this.isFilePart && this.partSize >= maxInMemorySize) {
+				throw new DataBufferLimitException("Part[" + this.index + "] " +
+						"exceeded the in-memory limit of " + maxInMemorySize + " bytes");
+			}
+			if (maxDiskUsagePerPart > 0 && this.isFilePart && this.partSize > maxDiskUsagePerPart) {
+				throw new DecodingException("Part[" + this.index + "] " +
+						"exceeded the disk usage limit of " + maxDiskUsagePerPart + " bytes");
+			}
+		}
+
+		public void partFinished() {
+			this.index++;
+			this.isFilePart = false;
+			this.partSize = 0;
+		}
+	}
+
+
 	/**
 	 * Listen for parser output and adapt to {@code Flux<Sink<Part>>}.
 	 */
@@ -193,43 +338,46 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 
 		private final FluxSink<Part> sink;
 
-		private final DataBufferFactory bufferFactory;
-
 		private final MultipartContext context;
 
+		private final LimitedPartBodyStreamStorageFactory storageFactory;
+
 		private final AtomicInteger terminated = new AtomicInteger(0);
 
-		FluxSinkAdapterListener(FluxSink<Part> sink, DataBufferFactory factory, MultipartContext context) {
+		FluxSinkAdapterListener(
+				FluxSink<Part> sink, MultipartContext context, LimitedPartBodyStreamStorageFactory factory) {
+
 			this.sink = sink;
-			this.bufferFactory = factory;
 			this.context = context;
+			this.storageFactory = factory;
 		}
 
 		@Override
 		public void onPartFinished(StreamStorage storage, Map<String, List<String>> headers) {
 			HttpHeaders httpHeaders = new HttpHeaders();
 			httpHeaders.putAll(headers);
+			this.storageFactory.partFinished();
 			this.sink.next(createPart(storage, httpHeaders));
 		}
 
 		private Part createPart(StreamStorage storage, HttpHeaders httpHeaders) {
 			String filename = MultipartUtils.getFileName(httpHeaders);
 			if (filename != null) {
-				return new SynchronossFilePart(httpHeaders, filename, storage, this.bufferFactory);
+				return new SynchronossFilePart(httpHeaders, filename, storage);
 			}
 			else if (MultipartUtils.isFormField(httpHeaders, this.context)) {
 				String value = MultipartUtils.readFormParameterValue(storage, httpHeaders);
-				return new SynchronossFormFieldPart(httpHeaders, this.bufferFactory, value);
+				return new SynchronossFormFieldPart(httpHeaders, value);
 			}
 			else {
-				return new SynchronossPart(httpHeaders, storage, this.bufferFactory);
+				return new SynchronossPart(httpHeaders, storage);
 			}
 		}
 
 		@Override
 		public void onError(String message, Throwable cause) {
 			if (this.terminated.getAndIncrement() == 0) {
-				this.sink.error(new RuntimeException(message, cause));
+				this.sink.error(new DecodingException(message, cause));
 			}
 		}
 
@@ -256,14 +404,10 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 
 		private final HttpHeaders headers;
 
-		private final DataBufferFactory bufferFactory;
-
-		AbstractSynchronossPart(HttpHeaders headers, DataBufferFactory bufferFactory) {
+		AbstractSynchronossPart(HttpHeaders headers) {
 			Assert.notNull(headers, "HttpHeaders is required");
-			Assert.notNull(bufferFactory, "DataBufferFactory is required");
 			this.name = MultipartUtils.getFieldName(headers);
 			this.headers = headers;
-			this.bufferFactory = bufferFactory;
 		}
 
 		@Override
@@ -276,10 +420,6 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 			return this.headers;
 		}
 
-		DataBufferFactory getBufferFactory() {
-			return this.bufferFactory;
-		}
-
 		@Override
 		public String toString() {
 			return "Part '" + this.name + "', headers=" + this.headers;
@@ -291,15 +431,15 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 
 		private final StreamStorage storage;
 
-		SynchronossPart(HttpHeaders headers, StreamStorage storage, DataBufferFactory factory) {
-			super(headers, factory);
+		SynchronossPart(HttpHeaders headers, StreamStorage storage) {
+			super(headers);
 			Assert.notNull(storage, "StreamStorage is required");
 			this.storage = storage;
 		}
 
 		@Override
 		public Flux<DataBuffer> content() {
-			return DataBufferUtils.readInputStream(getStorage()::getInputStream, getBufferFactory(), 4096);
+			return DataBufferUtils.readInputStream(getStorage()::getInputStream, bufferFactory, 4096);
 		}
 
 		protected StreamStorage getStorage() {
@@ -315,8 +455,8 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 
 		private final String filename;
 
-		SynchronossFilePart(HttpHeaders headers, String filename, StreamStorage storage, DataBufferFactory factory) {
-			super(headers, storage, factory);
+		SynchronossFilePart(HttpHeaders headers, String filename, StreamStorage storage) {
+			super(headers, storage);
 			this.filename = filename;
 		}
 
@@ -375,8 +515,8 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 
 		private final String content;
 
-		SynchronossFormFieldPart(HttpHeaders headers, DataBufferFactory bufferFactory, String content) {
-			super(headers, bufferFactory);
+		SynchronossFormFieldPart(HttpHeaders headers, String content) {
+			super(headers);
 			this.content = content;
 		}
 
@@ -388,9 +528,7 @@ public class SynchronossPartHttpMessageReader extends LoggingCodecSupport implem
 		@Override
 		public Flux<DataBuffer> content() {
 			byte[] bytes = this.content.getBytes(getCharset());
-			DataBuffer buffer = getBufferFactory().allocateBuffer(bytes.length);
-			buffer.write(bytes);
-			return Flux.just(buffer);
+			return Flux.just(bufferFactory.wrap(bytes));
 		}
 
 		private Charset getCharset() {

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

@@ -36,6 +36,7 @@ import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.Decoder;
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -101,10 +102,24 @@ public class ProtobufDecoder extends ProtobufCodecSupport implements Decoder<Mes
 	}
 
 
+	/**
+	 * The max size allowed per message.
+	 * <p>By default in 5.1 this is set to 64K. In 5.2 the default for this limit
+	 * 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) {
@@ -127,7 +142,7 @@ public class ProtobufDecoder extends ProtobufCodecSupport implements Decoder<Mes
 	public Mono<Message> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
-		return DataBufferUtils.join(inputStream).map(dataBuffer -> {
+		return DataBufferUtils.join(inputStream, getMaxMessageSize()).map(dataBuffer -> {
 					try {
 						Message.Builder builder = getMessageBuilder(elementType.toClass());
 						ByteBuffer buffer = dataBuffer.asByteBuffer();
@@ -198,9 +213,9 @@ public class ProtobufDecoder extends ProtobufCodecSupport implements Decoder<Mes
 						if (!readMessageSize(input)) {
 							return messages;
 						}
-						if (this.messageBytesToRead > this.maxMessageSize) {
-							throw new DecodingException(
-									"The number of bytes to read from the incoming stream " +
+						if (this.maxMessageSize > 0 && this.messageBytesToRead > this.maxMessageSize) {
+							throw new DataBufferLimitException(
+									"The number of bytes to read for message " +
 											"(" + this.messageBytesToRead + ") exceeds " +
 											"the configured limit (" + this.maxMessageSize + ")");
 						}

spring-web/src/main/java/org/springframework/http/codec/support/BaseDefaultCodecs.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.
@@ -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;
@@ -38,6 +39,7 @@ import org.springframework.http.codec.FormHttpMessageReader;
 import org.springframework.http.codec.HttpMessageReader;
 import org.springframework.http.codec.HttpMessageWriter;
 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 +97,9 @@ class BaseDefaultCodecs implements CodecConfigurer.DefaultCodecs {
 	@Nullable
 	private Encoder<?> jaxb2Encoder;
 
+	@Nullable
+	private Integer maxInMemorySize;
+
 	private boolean enableLoggingRequestDetails = false;
 
 	private boolean registerDefaults = true;
@@ -130,6 +135,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 +170,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 DecoderHttpMessageReader<>(new ResourceDecoder()));
-		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 DecoderHttpMessageReader<>(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 +192,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 +229,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 +256,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

@@ -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.
@@ -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/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java

@@ -43,6 +43,7 @@ import org.springframework.core.codec.CodecException;
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.codec.Hints;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.log.LogFormatUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
@@ -78,6 +79,8 @@ public class Jaxb2XmlDecoder extends AbstractDecoder<Object> {
 
 	private Function<Unmarshaller, Unmarshaller> unmarshallerProcessor = Function.identity();
 
+	private int maxInMemorySize = -1;
+
 
 	public Jaxb2XmlDecoder() {
 		super(MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML);
@@ -110,6 +113,29 @@ public class Jaxb2XmlDecoder extends AbstractDecoder<Object> {
 		return this.unmarshallerProcessor;
 	}
 
+	/**
+	 * Set the max number of bytes that can be buffered by this decoder.
+	 * This is either the size of the entire input when decoding as a whole, or when
+	 * using async parsing with Aalto XML, it is the size of one top-level XML tree.
+	 * When the limit is exceeded, {@link DataBufferLimitException} is raised.
+	 * <p>By default in 5.1 this is set to -1, unlimited. In 5.2 the default
+	 * value for this limit is set to 256K.
+	 * @param byteCount the max number of bytes to buffer, or -1 for unlimited
+	 * @since 5.1.11
+	 */
+	public void setMaxInMemorySize(int byteCount) {
+		this.maxInMemorySize = byteCount;
+		this.xmlEventDecoder.setMaxInMemorySize(byteCount);
+	}
+
+	/**
+	 * Return the {@link #setMaxInMemorySize configured} byte count limit.
+	 * @since 5.1.11
+	 */
+	public int getMaxInMemorySize() {
+		return this.maxInMemorySize;
+	}
+
 
 	@Override
 	public boolean canDecode(ResolvableType elementType, @Nullable MimeType mimeType) {
@@ -166,7 +192,7 @@ public class Jaxb2XmlDecoder extends AbstractDecoder<Object> {
 		}
 	}
 
-	private Unmarshaller initUnmarshaller(Class<?> outputClass) throws JAXBException {
+	private Unmarshaller initUnmarshaller(Class<?> outputClass) throws CodecException, JAXBException {
 		Unmarshaller unmarshaller = this.jaxbContexts.createUnmarshaller(outputClass);
 		return this.unmarshallerProcessor.apply(unmarshaller);
 	}

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

@@ -135,7 +135,7 @@ public class Jaxb2XmlEncoder extends AbstractSingleValueEncoder<Object> {
 		}).flux();
 	}
 
-	private Marshaller initMarshaller(Class<?> clazz) throws JAXBException {
+	private Marshaller initMarshaller(Class<?> clazz) throws CodecException, JAXBException {
 		Marshaller marshaller = this.jaxbContexts.createMarshaller(clazz);
 		marshaller.setProperty(Marshaller.JAXB_ENCODING, StandardCharsets.UTF_8.name());
 		marshaller = this.marshallerProcessor.apply(marshaller);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 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.
@@ -24,12 +24,13 @@ import javax.xml.bind.JAXBException;
 import javax.xml.bind.Marshaller;
 import javax.xml.bind.Unmarshaller;
 
-import org.springframework.util.Assert;
+import org.springframework.core.codec.CodecException;
 
 /**
  * Holder for {@link JAXBContext} instances.
  *
  * @author Arjen Poutsma
+ * @author Juergen Hoeller
  * @since 5.0
  */
 final class JaxbContextContainer {
@@ -37,24 +38,26 @@ final class JaxbContextContainer {
 	private final ConcurrentMap<Class<?>, JAXBContext> jaxbContexts = new ConcurrentHashMap<>(64);
 
 
-	public Marshaller createMarshaller(Class<?> clazz) throws JAXBException {
+	public Marshaller createMarshaller(Class<?> clazz) throws CodecException, JAXBException {
 		JAXBContext jaxbContext = getJaxbContext(clazz);
 		return jaxbContext.createMarshaller();
 	}
 
-	public Unmarshaller createUnmarshaller(Class<?> clazz) throws JAXBException {
+	public Unmarshaller createUnmarshaller(Class<?> clazz) throws CodecException, JAXBException {
 		JAXBContext jaxbContext = getJaxbContext(clazz);
 		return jaxbContext.createUnmarshaller();
 	}
 
-	private JAXBContext getJaxbContext(Class<?> clazz) throws JAXBException {
-		Assert.notNull(clazz, "Class must not be null");
-		JAXBContext jaxbContext = this.jaxbContexts.get(clazz);
-		if (jaxbContext == null) {
-			jaxbContext = JAXBContext.newInstance(clazz);
-			this.jaxbContexts.putIfAbsent(clazz, jaxbContext);
-		}
-		return jaxbContext;
+	private JAXBContext getJaxbContext(Class<?> clazz) throws CodecException {
+		return this.jaxbContexts.computeIfAbsent(clazz, key -> {
+			try {
+				return JAXBContext.newInstance(clazz);
+			}
+			catch (JAXBException ex) {
+				throw new CodecException(
+						"Could not create JAXBContext for class [" + clazz + "]: " + ex.getMessage(), ex);
+			}
+		});
 	}
 
 }

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

@@ -40,6 +40,7 @@ import reactor.core.publisher.Flux;
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.AbstractDecoder;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.ClassUtils;
@@ -88,26 +89,51 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
 
 	boolean useAalto = aaltoPresent;
 
+	private int maxInMemorySize = -1;
+
 
 	public XmlEventDecoder() {
 		super(MimeTypeUtils.APPLICATION_XML, MimeTypeUtils.TEXT_XML);
 	}
 
 
+	/**
+	 * Set the max number of bytes that can be buffered by this decoder. This
+	 * is either the size the entire input when decoding as a whole, or when
+	 * using async parsing via Aalto XML, it is size one top-level XML tree.
+	 * When the limit is exceeded, {@link DataBufferLimitException} is raised.
+	 * <p>By default in 5.1 this is set to -1, unlimited. In 5.2 the default
+	 * value for this limit is set to 256K.
+	 * @param byteCount the max number of bytes to buffer, or -1 for unlimited
+	 * @since 5.1.11
+	 */
+	public void setMaxInMemorySize(int byteCount) {
+		this.maxInMemorySize = byteCount;
+	}
+
+	/**
+	 * Return the {@link #setMaxInMemorySize configured} byte count limit.
+	 * @since 5.1.11
+	 */
+	public int getMaxInMemorySize() {
+		return this.maxInMemorySize;
+	}
+
+
 	@Override
 	@SuppressWarnings({"rawtypes", "unchecked"})  // on JDK 9 where XMLEventReader is Iterator<Object>
 	public Flux<XMLEvent> decode(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
 		if (this.useAalto) {
-			AaltoDataBufferToXmlEvent mapper = new AaltoDataBufferToXmlEvent();
+			AaltoDataBufferToXmlEvent mapper = new AaltoDataBufferToXmlEvent(this.maxInMemorySize);
 			return Flux.from(input)
 					.flatMapIterable(mapper)
 					.doFinally(signalType -> mapper.endOfInput());
 		}
 		else {
-			return DataBufferUtils.join(input).
-					flatMapIterable(buffer -> {
+			return DataBufferUtils.join(input, getMaxInMemorySize())
+					.flatMapIterable(buffer -> {
 						try {
 							InputStream is = buffer.asInputStream();
 							Iterator eventReader = inputFactory.createXMLEventReader(is);
@@ -139,10 +165,22 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
 
 		private final XMLEventAllocator eventAllocator = EventAllocatorImpl.getDefaultInstance();
 
+		private final int maxInMemorySize;
+
+		private int byteCount;
+
+		private int elementDepth;
+
+
+		public AaltoDataBufferToXmlEvent(int maxInMemorySize) {
+			this.maxInMemorySize = maxInMemorySize;
+		}
+
 
 		@Override
 		public List<? extends XMLEvent> apply(DataBuffer dataBuffer) {
 			try {
+				increaseByteCount(dataBuffer);
 				this.streamReader.getInputFeeder().feedInput(dataBuffer.asByteBuffer());
 				List<XMLEvent> events = new ArrayList<>();
 				while (true) {
@@ -156,8 +194,12 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
 						if (event.isEndDocument()) {
 							break;
 						}
+						checkDepthAndResetByteCount(event);
 					}
 				}
+				if (this.maxInMemorySize > 0 && this.byteCount > this.maxInMemorySize) {
+					raiseLimitException();
+				}
 				return events;
 			}
 			catch (XMLStreamException ex) {
@@ -168,6 +210,35 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
 			}
 		}
 
+		private void increaseByteCount(DataBuffer dataBuffer) {
+			if (this.maxInMemorySize > 0) {
+				if (dataBuffer.readableByteCount() > Integer.MAX_VALUE - this.byteCount) {
+					raiseLimitException();
+				}
+				else {
+					this.byteCount += dataBuffer.readableByteCount();
+				}
+			}
+		}
+
+		private void checkDepthAndResetByteCount(XMLEvent event) {
+			if (this.maxInMemorySize > 0) {
+				if (event.isStartElement()) {
+					this.byteCount = this.elementDepth == 1 ? 0 : this.byteCount;
+					this.elementDepth++;
+				}
+				else if (event.isEndElement()) {
+					this.elementDepth--;
+					this.byteCount = this.elementDepth == 1 ? 0 : this.byteCount;
+				}
+			}
+		}
+
+		private void raiseLimitException() {
+			throw new DataBufferLimitException(
+					"Exceeded limit on max bytes per XML top-level node: " + this.maxInMemorySize);
+		}
+
 		public void endOfInput() {
 			this.streamReader.getInputFeeder().endOfInput();
 		}

spring-web/src/main/java/org/springframework/http/converter/xml/AbstractJaxb2HttpMessageConverter.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.
@@ -25,7 +25,6 @@ import javax.xml.bind.Marshaller;
 import javax.xml.bind.Unmarshaller;
 
 import org.springframework.http.converter.HttpMessageConversionException;
-import org.springframework.util.Assert;
 
 /**
  * Abstract base class for {@link org.springframework.http.converter.HttpMessageConverter HttpMessageConverters}
@@ -106,19 +105,15 @@ public abstract class AbstractJaxb2HttpMessageConverter<T> extends AbstractXmlHt
 	 * @throws HttpMessageConversionException in case of JAXB errors
 	 */
 	protected final JAXBContext getJaxbContext(Class<?> clazz) {
-		Assert.notNull(clazz, "Class must not be null");
-		JAXBContext jaxbContext = this.jaxbContexts.get(clazz);
-		if (jaxbContext == null) {
+		return this.jaxbContexts.computeIfAbsent(clazz, key -> {
 			try {
-				jaxbContext = JAXBContext.newInstance(clazz);
-				this.jaxbContexts.putIfAbsent(clazz, jaxbContext);
+				return JAXBContext.newInstance(clazz);
 			}
 			catch (JAXBException ex) {
 				throw new HttpMessageConversionException(
-						"Could not instantiate JAXBContext for class [" + clazz + "]: " + ex.getMessage(), ex);
+						"Could not create JAXBContext for class [" + clazz + "]: " + ex.getMessage(), ex);
 			}
-		}
-		return jaxbContext;
+		});
 	}
 
 }

spring-web/src/main/java/org/springframework/web/server/MethodNotAllowedException.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.
@@ -19,12 +19,15 @@ package org.springframework.web.server;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Exception for errors that fit response status 405 (method not allowed).
@@ -37,7 +40,7 @@ public class MethodNotAllowedException extends ResponseStatusException {
 
 	private final String method;
 
-	private final Set<HttpMethod> supportedMethods;
+	private final Set<HttpMethod> httpMethods;
 
 
 	public MethodNotAllowedException(HttpMethod method, Collection<HttpMethod> supportedMethods) {
@@ -51,10 +54,21 @@ public class MethodNotAllowedException extends ResponseStatusException {
 			supportedMethods = Collections.emptySet();
 		}
 		this.method = method;
-		this.supportedMethods = Collections.unmodifiableSet(new HashSet<>(supportedMethods));
+		this.httpMethods = Collections.unmodifiableSet(new HashSet<>(supportedMethods));
 	}
 
 
+	/**
+	 * Return a Map with an "Allow" header.
+	 * @since 5.1.11
+	 */
+	@Override
+	public Map<String, String> getHeaders() {
+		return !CollectionUtils.isEmpty(this.httpMethods) ?
+				Collections.singletonMap("Allow", StringUtils.collectionToDelimitedString(this.httpMethods, ", ")) :
+				Collections.emptyMap();
+	}
+
 	/**
 	 * Return the HTTP method for the failed request.
 	 */
@@ -66,6 +80,7 @@ public class MethodNotAllowedException extends ResponseStatusException {
 	 * Return the list of supported HTTP methods.
 	 */
 	public Set<HttpMethod> getSupportedMethods() {
-		return this.supportedMethods;
+		return this.httpMethods;
 	}
+
 }

spring-web/src/main/java/org/springframework/web/server/NotAcceptableStatusException.java

@@ -18,9 +18,11 @@ package org.springframework.web.server;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
+import org.springframework.util.CollectionUtils;
 
 /**
  * Exception for errors that fit response status 406 (not acceptable).
@@ -51,6 +53,17 @@ public class NotAcceptableStatusException extends ResponseStatusException {
 	}
 
 
+	/**
+	 * Return a Map with an "Accept" header, or an empty map.
+	 * @since 5.1.11
+	 */
+	@Override
+	public Map<String, String> getHeaders() {
+		return !CollectionUtils.isEmpty(this.supportedMediaTypes) ?
+				Collections.singletonMap("Accept", MediaType.toString(this.supportedMediaTypes)) :
+				Collections.emptyMap();
+	}
+
 	/**
 	 * Return the list of supported content types in cases when the Accept
 	 * header is parsed but not supported, or an empty list otherwise.

spring-web/src/main/java/org/springframework/web/server/ResponseStatusException.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.
@@ -16,6 +16,9 @@
 
 package org.springframework.web.server;
 
+import java.util.Collections;
+import java.util.Map;
+
 import org.springframework.core.NestedExceptionUtils;
 import org.springframework.core.NestedRuntimeException;
 import org.springframework.http.HttpStatus;
@@ -72,12 +75,21 @@ public class ResponseStatusException extends NestedRuntimeException {
 
 
 	/**
-	 * The HTTP status that fits the exception (never {@code null}).
+	 * Return the HTTP status associated with this exception.
 	 */
 	public HttpStatus getStatus() {
 		return this.status;
 	}
 
+	/**
+	 * Return response headers associated with the exception, possibly required
+	 * for the given status code (e.g. "Allow", "Accept").
+	 * @since 5.1.11
+	 */
+	public Map<String, String> getHeaders() {
+		return Collections.emptyMap();
+	}
+
 	/**
 	 * The reason explaining the exception (potentially {@code null} or empty).
 	 */
@@ -86,6 +98,7 @@ public class ResponseStatusException extends NestedRuntimeException {
 		return this.reason;
 	}
 
+
 	@Override
 	public String getMessage() {
 		String msg = this.status + (this.reason != null ? " \"" + this.reason + "\"" : "");

spring-web/src/main/java/org/springframework/web/server/handler/ResponseStatusExceptionHandler.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.
@@ -22,6 +22,7 @@ import reactor.core.publisher.Mono;
 
 import org.springframework.http.HttpStatus;
 import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.http.server.reactive.ServerHttpResponse;
 import org.springframework.lang.Nullable;
 import org.springframework.web.server.ResponseStatusException;
 import org.springframework.web.server.ServerWebExchange;
@@ -62,8 +63,7 @@ public class ResponseStatusExceptionHandler implements WebExceptionHandler {
 
 	@Override
 	public Mono<Void> handle(ServerWebExchange exchange, Throwable ex) {
-		HttpStatus status = resolveStatus(ex);
-		if (status == null || !exchange.getResponse().setStatusCode(status)) {
+		if (!updateResponse(exchange.getResponse(), ex)) {
 			return Mono.error(ex);
 		}
 
@@ -86,16 +86,25 @@ public class ResponseStatusExceptionHandler implements WebExceptionHandler {
 		return "Resolved [" + reason + "] for HTTP " + request.getMethod() + " " + path;
 	}
 
-	@Nullable
-	private HttpStatus resolveStatus(Throwable ex) {
+	private boolean updateResponse(ServerHttpResponse response, Throwable ex) {
+		boolean result = false;
 		HttpStatus status = determineStatus(ex);
-		if (status == null) {
-			Throwable cause = ex.getCause();
-			if (cause != null) {
-				status = resolveStatus(cause);
+		if (status != null) {
+			if (response.setStatusCode(status)) {
+				if (ex instanceof ResponseStatusException) {
+					((ResponseStatusException) ex).getHeaders()
+							.forEach((name, value) -> response.getHeaders().add(name, value));
+				}
+				result = true;
 			}
 		}
-		return status;
+		else {
+			Throwable cause = ex.getCause();
+			if (cause != null) {
+				result = updateResponse(response, cause);
+			}
+		}
+		return result;
 	}
 
 	/**

spring-web/src/test/java/org/springframework/http/codec/json/Jackson2TokenizerTests.java

@@ -20,7 +20,6 @@ import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
-import java.util.function.Consumer;
 
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.TreeNode;
@@ -36,6 +35,7 @@ import reactor.test.StepVerifier;
 import org.springframework.core.codec.DecodingException;
 import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 
 import static java.util.Arrays.*;
 import static java.util.Collections.*;
@@ -181,11 +181,68 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 		testTokenize(asList("[1", ",2,", "3]"), asList("1", "2", "3"), true);
 	}
 
+	private void testTokenize(List<String> input, List<String> output, boolean tokenize) {
+		StepVerifier.FirstStep<String> builder = StepVerifier.create(decode(input, tokenize, -1));
+		output.forEach(expected -> builder.assertNext(actual -> {
+			try {
+				JSONAssert.assertEquals(expected, actual, true);
+			}
+			catch (JSONException ex) {
+				throw new RuntimeException(ex);
+			}
+		}));
+		builder.verifyComplete();
+	}
+
+	@Test
+	public void testLimit() {
+
+		List<String> source = asList("[",
+				"{", "\"id\":1,\"name\":\"Dan\"", "},",
+				"{", "\"id\":2,\"name\":\"Ron\"", "},",
+				"{", "\"id\":3,\"name\":\"Bartholomew\"", "}",
+				"]");
+
+		String expected = String.join("", source);
+		int maxInMemorySize = expected.length();
+
+		StepVerifier.create(decode(source, false, maxInMemorySize))
+				.expectNext(expected)
+				.verifyComplete();
+
+		StepVerifier.create(decode(source, false, maxInMemorySize - 1))
+				.expectError(DataBufferLimitException.class);
+	}
+
+	@Test
+	public void testLimitTokenized() {
+
+		List<String> source = asList("[",
+				"{", "\"id\":1, \"name\":\"Dan\"", "},",
+				"{", "\"id\":2, \"name\":\"Ron\"", "},",
+				"{", "\"id\":3, \"name\":\"Bartholomew\"", "}",
+				"]");
+
+		String expected = "{\"id\":3,\"name\":\"Bartholomew\"}";
+		int maxInMemorySize = expected.length();
+
+		StepVerifier.create(decode(source, true, maxInMemorySize))
+				.expectNext("{\"id\":1,\"name\":\"Dan\"}")
+				.expectNext("{\"id\":2,\"name\":\"Ron\"}")
+				.expectNext(expected)
+				.verifyComplete();
+
+		StepVerifier.create(decode(source, true, maxInMemorySize - 1))
+				.expectNext("{\"id\":1,\"name\":\"Dan\"}")
+				.expectNext("{\"id\":2,\"name\":\"Ron\"}")
+				.verifyError(DataBufferLimitException.class);
+	}
+
 	@Test
 	public void errorInStream() {
 		DataBuffer buffer = stringBuffer("{\"id\":1,\"name\":");
 		Flux<DataBuffer> source = Flux.just(buffer).concatWith(Flux.error(new RuntimeException()));
-		Flux<TokenBuffer> result = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, true);
+		Flux<TokenBuffer> result = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, true, -1);
 
 		StepVerifier.create(result)
 				.expectError(RuntimeException.class)
@@ -195,7 +252,7 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 	@Test  // SPR-16521
 	public void jsonEOFExceptionIsWrappedAsDecodingError() {
 		Flux<DataBuffer> source = Flux.just(stringBuffer("{\"status\": \"noClosingQuote}"));
-		Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, false);
+		Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(source, this.jsonFactory, this.objectMapper, false, -1);
 
 		StepVerifier.create(tokens)
 				.expectError(DecodingException.class)
@@ -203,12 +260,13 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 	}
 
 
-	private void testTokenize(List<String> source, List<String> expected, boolean tokenizeArrayElements) {
+	private Flux<String> decode(List<String> source, boolean tokenize, int maxInMemorySize) {
+
 		Flux<TokenBuffer> tokens = Jackson2Tokenizer.tokenize(
 				Flux.fromIterable(source).map(this::stringBuffer),
-				this.jsonFactory, this.objectMapper, tokenizeArrayElements);
+				this.jsonFactory, this.objectMapper, tokenize, maxInMemorySize);
 
-		Flux<String> result = tokens
+		return tokens
 				.map(tokenBuffer -> {
 					try {
 						TreeNode root = this.objectMapper.readTree(tokenBuffer.asParser());
@@ -218,10 +276,6 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 						throw new UncheckedIOException(ex);
 					}
 				});
-
-		StepVerifier.FirstStep<String> builder = StepVerifier.create(result);
-		expected.forEach(s -> builder.assertNext(new JSONAssertConsumer(s)));
-		builder.verifyComplete();
 	}
 
 	private DataBuffer stringBuffer(String value) {
@@ -231,24 +285,4 @@ public class Jackson2TokenizerTests extends AbstractLeakCheckingTestCase {
 		return buffer;
 	}
 
-
-	private static class JSONAssertConsumer implements Consumer<String> {
-
-		private final String expected;
-
-		JSONAssertConsumer(String expected) {
-			this.expected = expected;
-		}
-
-		@Override
-		public void accept(String s) {
-			try {
-				JSONAssert.assertEquals(this.expected, s, true);
-			}
-			catch (JSONException ex) {
-				throw new RuntimeException(ex);
-			}
-		}
-	}
-
 }

spring-web/src/test/java/org/springframework/http/codec/multipart/SynchronossPartHttpMessageReaderTests.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.
@@ -17,15 +17,20 @@
 package org.springframework.http.codec.multipart;
 
 import java.io.File;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.Map;
+import java.util.function.Consumer;
 
 import org.junit.Test;
+import org.reactivestreams.Subscription;
+import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.core.ResolvableType;
+import org.springframework.core.codec.DecodingException;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
@@ -38,23 +43,31 @@ import org.springframework.mock.http.client.reactive.test.MockClientHttpRequest;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
 import org.springframework.util.MultiValueMap;
 
-import static java.util.Collections.*;
-import static org.junit.Assert.*;
-import static org.springframework.core.ResolvableType.*;
-import static org.springframework.http.HttpHeaders.*;
-import static org.springframework.http.MediaType.*;
+import static java.util.Collections.emptyMap;
+import static org.hamcrest.core.StringStartsWith.startsWith;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.springframework.core.ResolvableType.forClassWithGenerics;
+import static org.springframework.http.HttpHeaders.CONTENT_TYPE;
+import static org.springframework.http.MediaType.MULTIPART_FORM_DATA;
 
 /**
  * Unit tests for {@link SynchronossPartHttpMessageReader}.
  *
  * @author Sebastien Deleuze
  * @author Rossen Stoyanchev
+ * @author Brian Clozel
  */
 public class SynchronossPartHttpMessageReaderTests {
 
 	private final MultipartHttpMessageReader reader =
 			new MultipartHttpMessageReader(new SynchronossPartHttpMessageReader());
 
+	private static final ResolvableType PARTS_ELEMENT_TYPE =
+			forClassWithGenerics(MultiValueMap.class, String.class, Part.class);
 
 	@Test
 	public void canRead() {
@@ -86,10 +99,10 @@ public class SynchronossPartHttpMessageReaderTests {
 		MultiValueMap<String, Part> parts = this.reader.readMono(elementType, request, emptyMap()).block();
 		assertEquals(2, parts.size());
 
-		assertTrue(parts.containsKey("fooPart"));
-		Part part = parts.getFirst("fooPart");
+		assertTrue(parts.containsKey("filePart"));
+		Part part = parts.getFirst("filePart");
 		assertTrue(part instanceof FilePart);
-		assertEquals("fooPart", part.name());
+		assertEquals("filePart", part.name());
 		assertEquals("foo.txt", ((FilePart) part).filename());
 		DataBuffer buffer = DataBufferUtils.join(part.content()).block();
 		assertEquals(12, buffer.readableByteCount());
@@ -97,24 +110,23 @@ public class SynchronossPartHttpMessageReaderTests {
 		buffer.read(byteContent);
 		assertEquals("Lorem Ipsum.", new String(byteContent));
 
-		assertTrue(parts.containsKey("barPart"));
-		part = parts.getFirst("barPart");
+		assertTrue(parts.containsKey("textPart"));
+		part = parts.getFirst("textPart");
 		assertTrue(part instanceof FormFieldPart);
-		assertEquals("barPart", part.name());
-		assertEquals("bar", ((FormFieldPart) part).value());
+		assertEquals("textPart", part.name());
+		assertEquals("sample-text", ((FormFieldPart) part).value());
 	}
 
 	@Test // SPR-16545
-	public void transferTo() {
+	public void transferTo() throws IOException {
 		ServerHttpRequest request = generateMultipartRequest();
-		ResolvableType elementType = forClassWithGenerics(MultiValueMap.class, String.class, Part.class);
-		MultiValueMap<String, Part> parts = this.reader.readMono(elementType, request, emptyMap()).block();
+		MultiValueMap<String, Part> parts = this.reader.readMono(PARTS_ELEMENT_TYPE, request, emptyMap()).block();
 
 		assertNotNull(parts);
-		FilePart part = (FilePart) parts.getFirst("fooPart");
+		FilePart part = (FilePart) parts.getFirst("filePart");
 		assertNotNull(part);
 
-		File dest = new File(System.getProperty("java.io.tmpdir") + "/" + part.filename());
+		File dest = File.createTempFile(part.filename(), "multipart");
 		part.transferTo(dest).block(Duration.ofSeconds(5));
 
 		assertTrue(dest.exists());
@@ -125,22 +137,65 @@ public class SynchronossPartHttpMessageReaderTests {
 	@Test
 	public void bodyError() {
 		ServerHttpRequest request = generateErrorMultipartRequest();
-		ResolvableType elementType = forClassWithGenerics(MultiValueMap.class, String.class, Part.class);
-		StepVerifier.create(this.reader.readMono(elementType, request, emptyMap())).verifyError();
+		StepVerifier.create(this.reader.readMono(PARTS_ELEMENT_TYPE, request, emptyMap())).verifyError();
 	}
 
+	@Test
+	public void readPartsWithoutDemand() {
+		ServerHttpRequest request = generateMultipartRequest();
+		Mono<MultiValueMap<String, Part>> parts = this.reader.readMono(PARTS_ELEMENT_TYPE, request, emptyMap());
+		ZeroDemandSubscriber subscriber = new ZeroDemandSubscriber();
+		parts.subscribe(subscriber);
+		subscriber.cancel();
+	}
+
+	@Test
+	public void readTooManyParts() {
+		testMultipartExceptions(reader -> reader.setMaxParts(1), ex -> {
+			assertEquals(DecodingException.class, ex.getClass());
+			assertThat(ex.getMessage(), startsWith("Failure while parsing part[2]"));
+			assertEquals("Too many parts (2 allowed)", ex.getCause().getMessage());
+		});
+	}
+
+	@Test
+	public void readFilePartTooBig() {
+		testMultipartExceptions(reader -> reader.setMaxDiskUsagePerPart(5), ex -> {
+			assertEquals(DecodingException.class, ex.getClass());
+			assertThat(ex.getMessage(), startsWith("Failure while parsing part[1]"));
+			assertEquals("Part[1] exceeded the disk usage limit of 5 bytes", ex.getCause().getMessage());
+		});
+	}
+
+	@Test
+	public void readPartHeadersTooBig() {
+		testMultipartExceptions(reader -> reader.setMaxInMemorySize(1), ex -> {
+			assertEquals(DecodingException.class, ex.getClass());
+			assertThat(ex.getMessage(), startsWith("Failure while parsing part[1]"));
+			assertEquals("Part[1] exceeded the in-memory limit of 1 bytes", ex.getCause().getMessage());
+		});
+	}
+
+	private void testMultipartExceptions(
+			Consumer<SynchronossPartHttpMessageReader> configurer, Consumer<Throwable> assertions) {
+
+		SynchronossPartHttpMessageReader reader = new SynchronossPartHttpMessageReader();
+		configurer.accept(reader);
+		MultipartHttpMessageReader multipartReader = new MultipartHttpMessageReader(reader);
+		StepVerifier.create(multipartReader.readMono(PARTS_ELEMENT_TYPE, generateMultipartRequest(), emptyMap()))
+				.consumeErrorWith(assertions)
+				.verify();
+	}
 
 	private ServerHttpRequest generateMultipartRequest() {
-
 		MultipartBodyBuilder partsBuilder = new MultipartBodyBuilder();
-		partsBuilder.part("fooPart", new ClassPathResource("org/springframework/http/codec/multipart/foo.txt"));
-		partsBuilder.part("barPart", "bar");
+		partsBuilder.part("filePart", new ClassPathResource("org/springframework/http/codec/multipart/foo.txt"));
+		partsBuilder.part("textPart", "sample-text");
 
 		MockClientHttpRequest outputMessage = new MockClientHttpRequest(HttpMethod.POST, "/");
 		new MultipartHttpMessageWriter()
 				.write(Mono.just(partsBuilder.build()), null, MediaType.MULTIPART_FORM_DATA, outputMessage, null)
 				.block(Duration.ofSeconds(5));
-
 		return MockServerHttpRequest.post("/")
 				.contentType(outputMessage.getHeaders().getContentType())
 				.body(outputMessage.getBody());
@@ -152,4 +207,12 @@ public class SynchronossPartHttpMessageReaderTests {
 				.body(Flux.just(new DefaultDataBufferFactory().wrap("invalid content".getBytes())));
 	}
 
+	private static class ZeroDemandSubscriber extends BaseSubscriber<MultiValueMap<String, Part>> {
+
+		@Override
+		protected void hookOnSubscribe(Subscription subscription) {
+			// Just subscribe without requesting
+		}
+	}
+
 }

spring-web/src/test/java/org/springframework/http/codec/support/ServerCodecConfigurerTests.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.
@@ -124,13 +124,41 @@ public class ServerCodecConfigurerTests {
 				.filter(e -> e == encoder).orElse(null));
 	}
 
+	@Test
+	public void maxInMemorySize() {
+		int size = 99;
+		this.configurer.defaultCodecs().maxInMemorySize(size);
+		List<HttpMessageReader<?>> readers = this.configurer.getReaders();
+		assertEquals(13, readers.size());
+		assertEquals(size, ((ByteArrayDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((ByteBufferDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((DataBufferDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((ResourceDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((StringDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((ProtobufDecoder) getNextDecoder(readers)).getMaxMessageSize());
+		assertEquals(size, ((FormHttpMessageReader) nextReader(readers)).getMaxInMemorySize());
+		assertEquals(size, ((SynchronossPartHttpMessageReader) nextReader(readers)).getMaxInMemorySize());
+
+		MultipartHttpMessageReader multipartReader = (MultipartHttpMessageReader) nextReader(readers);
+		SynchronossPartHttpMessageReader reader = (SynchronossPartHttpMessageReader) multipartReader.getPartReader();
+		assertEquals(size, (reader).getMaxInMemorySize());
+
+		assertEquals(size, ((Jackson2JsonDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((Jackson2SmileDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((Jaxb2XmlDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+		assertEquals(size, ((StringDecoder) getNextDecoder(readers)).getMaxInMemorySize());
+	}
 
 	private Decoder<?> getNextDecoder(List<HttpMessageReader<?>> readers) {
-		HttpMessageReader<?> reader = readers.get(this.index.getAndIncrement());
+		HttpMessageReader<?> reader = nextReader(readers);
 		assertEquals(DecoderHttpMessageReader.class, reader.getClass());
 		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());
 		assertEquals(EncoderHttpMessageWriter.class, writer.getClass());

spring-web/src/test/java/org/springframework/http/codec/xml/XmlEventDecoderTests.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.
@@ -28,8 +28,10 @@ import reactor.test.StepVerifier;
 
 import org.springframework.core.io.buffer.AbstractLeakCheckingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferLimitException;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 /**
  * @author Arjen Poutsma
@@ -44,11 +46,12 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 
 	private XmlEventDecoder decoder = new XmlEventDecoder();
 
+
 	@Test
 	public void toXMLEventsAalto() {
 
 		Flux<XMLEvent> events =
-				this.decoder.decode(stringBuffer(XML), null, null, Collections.emptyMap());
+				this.decoder.decode(stringBufferMono(XML), null, null, Collections.emptyMap());
 
 		StepVerifier.create(events)
 				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
@@ -69,7 +72,7 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 		decoder.useAalto = false;
 
 		Flux<XMLEvent> events =
-				this.decoder.decode(stringBuffer(XML), null, null, Collections.emptyMap());
+				this.decoder.decode(stringBufferMono(XML), null, null, Collections.emptyMap());
 
 		StepVerifier.create(events)
 				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
@@ -86,10 +89,32 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 				.verify();
 	}
 
+	@Test
+	public void toXMLEventsWithLimit() {
+
+		this.decoder.setMaxInMemorySize(6);
+
+		Flux<String> source = Flux.just(
+				"<pojo>", "<foo>", "foofoo", "</foo>", "<bar>", "barbarbar", "</bar>", "</pojo>");
+
+		Flux<XMLEvent> events = this.decoder.decode(
+				source.map(this::stringBuffer), null, null, Collections.emptyMap());
+
+		StepVerifier.create(events)
+				.consumeNextWith(e -> assertTrue(e.isStartDocument()))
+				.consumeNextWith(e -> assertStartElement(e, "pojo"))
+				.consumeNextWith(e -> assertStartElement(e, "foo"))
+				.consumeNextWith(e -> assertCharacters(e, "foofoo"))
+				.consumeNextWith(e -> assertEndElement(e, "foo"))
+				.consumeNextWith(e -> assertStartElement(e, "bar"))
+				.expectError(DataBufferLimitException.class)
+				.verify();
+	}
+
 	@Test
 	public void decodeErrorAalto() {
 		Flux<DataBuffer> source = Flux.concat(
-				stringBuffer("<pojo>"),
+				stringBufferMono("<pojo>"),
 				Flux.error(new RuntimeException()));
 
 		Flux<XMLEvent> events =
@@ -107,7 +132,7 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 		decoder.useAalto = false;
 
 		Flux<DataBuffer> source = Flux.concat(
-				stringBuffer("<pojo>"),
+				stringBufferMono("<pojo>"),
 				Flux.error(new RuntimeException()));
 
 		Flux<XMLEvent> events =
@@ -133,13 +158,15 @@ public class XmlEventDecoderTests extends AbstractLeakCheckingTestCase {
 		assertEquals(expectedData, event.asCharacters().getData());
 	}
 
-	private Mono<DataBuffer> stringBuffer(String value) {
-		return Mono.defer(() -> {
-			byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
-			DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
-			buffer.write(bytes);
-			return Mono.just(buffer);
-		});
+	private DataBuffer stringBuffer(String value) {
+		byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
+		DataBuffer buffer = this.bufferFactory.allocateBuffer(bytes.length);
+		buffer.write(bytes);
+		return buffer;
+	}
+
+	private Mono<DataBuffer> stringBufferMono(String value) {
+		return Mono.defer(() -> Mono.just(stringBuffer(value)));
 	}
 
 }

spring-web/src/test/java/org/springframework/mock/web/test/MockCookie.java

@@ -16,6 +16,9 @@
 
 package org.springframework.mock.web.test;
 
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+
 import javax.servlet.http.Cookie;
 
 import org.springframework.lang.Nullable;
@@ -35,12 +38,15 @@ public class MockCookie extends Cookie {
 	private static final long serialVersionUID = 4312531139502726325L;
 
 
+	@Nullable
+	private ZonedDateTime expires;
+
 	@Nullable
 	private String sameSite;
 
 
 	/**
-	 * Constructor with the cookie name and value.
+	 * Construct a new {@link MockCookie} with the supplied name and value.
 	 * @param name the name
 	 * @param value the value
 	 * @see Cookie#Cookie(String, String)
@@ -49,12 +55,29 @@ public class MockCookie extends Cookie {
 		super(name, value);
 	}
 
+	/**
+	 * Set the "Expires" attribute for this cookie.
+	 * @since 5.1.11
+	 */
+	public void setExpires(@Nullable ZonedDateTime expires) {
+		this.expires = expires;
+	}
 
 	/**
-	 * Add the "SameSite" attribute to the cookie.
+	 * Get the "Expires" attribute for this cookie.
+	 * @since 5.1.11
+	 * @return the "Expires" attribute for this cookie, or {@code null} if not set
+	 */
+	@Nullable
+	public ZonedDateTime getExpires() {
+		return this.expires;
+	}
+
+	/**
+	 * Set the "SameSite" attribute for this cookie.
 	 * <p>This limits the scope of the cookie such that it will only be attached
-	 * to same site requests if {@code "Strict"} or cross-site requests if
-	 * {@code "Lax"}.
+	 * to same-site requests if the supplied value is {@code "Strict"} or cross-site
+	 * requests if the supplied value is {@code "Lax"}.
 	 * @see <a href="https://tools.ietf.org/html/draft-ietf-httpbis-rfc6265bis#section-4.1.2.7">RFC6265 bis</a>
 	 */
 	public void setSameSite(@Nullable String sameSite) {
@@ -62,7 +85,8 @@ public class MockCookie extends Cookie {
 	}
 
 	/**
-	 * Return the "SameSite" attribute, or {@code null} if not set.
+	 * Get the "SameSite" attribute for this cookie.
+	 * @return the "SameSite" attribute for this cookie, or {@code null} if not set
 	 */
 	@Nullable
 	public String getSameSite() {
@@ -71,7 +95,7 @@ public class MockCookie extends Cookie {
 
 
 	/**
-	 * Factory method that parses the value of a "Set-Cookie" header.
+	 * Factory method that parses the value of the supplied "Set-Cookie" header.
 	 * @param setCookieHeader the "Set-Cookie" value; never {@code null} or empty
 	 * @return the created cookie
 	 */
@@ -94,6 +118,10 @@ public class MockCookie extends Cookie {
 			else if (StringUtils.startsWithIgnoreCase(attribute, "Max-Age")) {
 				cookie.setMaxAge(Integer.parseInt(extractAttributeValue(attribute, setCookieHeader)));
 			}
+			else if (StringUtils.startsWithIgnoreCase(attribute, "Expires")) {
+				cookie.setExpires(ZonedDateTime.parse(extractAttributeValue(attribute, setCookieHeader),
+						DateTimeFormatter.RFC_1123_DATE_TIME));
+			}
 			else if (StringUtils.startsWithIgnoreCase(attribute, "Path")) {
 				cookie.setPath(extractAttributeValue(attribute, setCookieHeader));
 			}

spring-web/src/test/java/org/springframework/mock/web/test/MockHttpServletResponse.java

@@ -26,6 +26,8 @@ import java.io.Writer;
 import java.text.DateFormat;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -345,9 +347,15 @@ public class MockHttpServletResponse implements HttpServletResponse {
 		if (maxAge >= 0) {
 			buf.append("; Max-Age=").append(maxAge);
 			buf.append("; Expires=");
-			HttpHeaders headers = new HttpHeaders();
-			headers.setExpires(maxAge > 0 ? System.currentTimeMillis() + 1000L * maxAge : 0);
-			buf.append(headers.getFirst(HttpHeaders.EXPIRES));
+			ZonedDateTime expires = (cookie instanceof MockCookie ? ((MockCookie) cookie).getExpires() : null);
+			if (expires != null) {
+				buf.append(expires.format(DateTimeFormatter.RFC_1123_DATE_TIME));
+			}
+			else {
+				HttpHeaders headers = new HttpHeaders();
+				headers.setExpires(maxAge > 0 ? System.currentTimeMillis() + 1000L * maxAge : 0);
+				buf.append(headers.getFirst(HttpHeaders.EXPIRES));
+			}
 		}
 
 		if (cookie.getSecure()) {

spring-web/src/test/java/org/springframework/mock/web/test/MockServletContext.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.
@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.nio.file.InvalidPathException;
 import java.util.Collections;
 import java.util.Enumeration;
 import java.util.EventListener;
@@ -294,8 +295,10 @@ public class MockServletContext implements ServletContext {
 	@Nullable
 	public Set<String> getResourcePaths(String path) {
 		String actualPath = (path.endsWith("/") ? path : path + "/");
-		Resource resource = this.resourceLoader.getResource(getResourceLocation(actualPath));
+		String resourceLocation = getResourceLocation(actualPath);
+		Resource resource = null;
 		try {
+			resource = this.resourceLoader.getResource(resourceLocation);
 			File file = resource.getFile();
 			String[] fileList = file.list();
 			if (ObjectUtils.isEmpty(fileList)) {
@@ -311,9 +314,10 @@ public class MockServletContext implements ServletContext {
 			}
 			return resourcePaths;
 		}
-		catch (IOException ex) {
+		catch (InvalidPathException | IOException ex ) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("Could not get resource paths for " + resource, ex);
+				logger.warn("Could not get resource paths for " +
+						(resource != null ? resource : resourceLocation), ex);
 			}
 			return null;
 		}
@@ -322,19 +326,22 @@ public class MockServletContext implements ServletContext {
 	@Override
 	@Nullable
 	public URL getResource(String path) throws MalformedURLException {
-		Resource resource = this.resourceLoader.getResource(getResourceLocation(path));
-		if (!resource.exists()) {
-			return null;
-		}
+		String resourceLocation = getResourceLocation(path);
+		Resource resource = null;
 		try {
+			resource = this.resourceLoader.getResource(resourceLocation);
+			if (!resource.exists()) {
+				return null;
+			}
 			return resource.getURL();
 		}
 		catch (MalformedURLException ex) {
 			throw ex;
 		}
-		catch (IOException ex) {
+		catch (InvalidPathException | IOException ex) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("Could not get URL for " + resource, ex);
+				logger.warn("Could not get URL for resource " +
+						(resource != null ? resource : resourceLocation), ex);
 			}
 			return null;
 		}
@@ -343,16 +350,19 @@ public class MockServletContext implements ServletContext {
 	@Override
 	@Nullable
 	public InputStream getResourceAsStream(String path) {
-		Resource resource = this.resourceLoader.getResource(getResourceLocation(path));
-		if (!resource.exists()) {
-			return null;
-		}
+		String resourceLocation = getResourceLocation(path);
+		Resource resource = null;
 		try {
+			resource = this.resourceLoader.getResource(resourceLocation);
+			if (!resource.exists()) {
+				return null;
+			}
 			return resource.getInputStream();
 		}
-		catch (IOException ex) {
+		catch (InvalidPathException | IOException ex) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("Could not open InputStream for " + resource, ex);
+				logger.warn("Could not open InputStream for resource " +
+						(resource != null ? resource : resourceLocation), ex);
 			}
 			return null;
 		}
@@ -459,13 +469,16 @@ public class MockServletContext implements ServletContext {
 	@Override
 	@Nullable
 	public String getRealPath(String path) {
-		Resource resource = this.resourceLoader.getResource(getResourceLocation(path));
+		String resourceLocation = getResourceLocation(path);
+		Resource resource = null;
 		try {
+			resource = this.resourceLoader.getResource(resourceLocation);
 			return resource.getFile().getAbsolutePath();
 		}
-		catch (IOException ex) {
+		catch (InvalidPathException | IOException ex) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("Could not determine real path of resource " + resource, ex);
+				logger.warn("Could not determine real path of resource " +
+						(resource != null ? resource : resourceLocation), ex);
 			}
 			return null;
 		}

spring-web/src/test/java/org/springframework/web/server/handler/ResponseStatusExceptionHandlerTests.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.
@@ -17,18 +17,27 @@
 package org.springframework.web.server.handler;
 
 import java.time.Duration;
+import java.util.Arrays;
 
 import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
+import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
 import org.springframework.mock.web.test.server.MockServerWebExchange;
+import org.springframework.web.server.MethodNotAllowedException;
+import org.springframework.web.server.NotAcceptableStatusException;
 import org.springframework.web.server.ResponseStatusException;
 
-import static org.junit.Assert.*;
+import static org.hamcrest.collection.IsIterableContainingInOrder.contains;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 
 /**
  * Unit tests for {@link ResponseStatusExceptionHandler}.
@@ -67,6 +76,26 @@ public class ResponseStatusExceptionHandlerTests {
 		assertEquals(HttpStatus.BAD_REQUEST, this.exchange.getResponse().getStatusCode());
 	}
 
+	@Test // gh-23741
+	public void handleMethodNotAllowed() {
+		Throwable ex = new MethodNotAllowedException(HttpMethod.PATCH, Arrays.asList(HttpMethod.POST, HttpMethod.PUT));
+		this.handler.handle(this.exchange, ex).block(Duration.ofSeconds(5));
+
+		MockServerHttpResponse response = this.exchange.getResponse();
+		assertEquals(HttpStatus.METHOD_NOT_ALLOWED, response.getStatusCode());
+		assertThat(response.getHeaders().getAllow(), contains(HttpMethod.POST, HttpMethod.PUT));
+	}
+
+	@Test // gh-23741
+	public void handleResponseStatusExceptionWithHeaders() {
+		Throwable ex = new NotAcceptableStatusException(Arrays.asList(MediaType.TEXT_PLAIN, MediaType.TEXT_HTML));
+		this.handler.handle(this.exchange, ex).block(Duration.ofSeconds(5));
+
+		MockServerHttpResponse response = this.exchange.getResponse();
+		assertEquals(HttpStatus.NOT_ACCEPTABLE, response.getStatusCode());
+		assertThat(response.getHeaders().getAccept(), contains(MediaType.TEXT_PLAIN, MediaType.TEXT_HTML));
+	}
+
 	@Test
 	public void unresolvedException() {
 		Throwable expected = new IllegalStateException();

spring-webmvc/spring-webmvc.gradle

@@ -24,7 +24,7 @@ dependencies {
 	optional("org.webjars:webjars-locator-core:0.37")
 	optional("com.rometools:rome:1.12.2")
 	optional("com.github.librepdf:openpdf:1.2.21")
-	optional("org.apache.poi:poi-ooxml:4.1.0")
+	optional("org.apache.poi:poi-ooxml:4.1.1")
 	optional("org.freemarker:freemarker:${freemarkerVersion}")
 	optional("com.fasterxml.jackson.core:jackson-databind:${jackson2Version}")
 	optional("com.fasterxml.jackson.dataformat:jackson-dataformat-xml:${jackson2Version}")

spring-websocket/src/main/java/org/springframework/web/socket/messaging/WebSocketStompClient.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.
@@ -57,6 +57,7 @@ import org.springframework.web.socket.WebSocketHttpHeaders;
 import org.springframework.web.socket.WebSocketMessage;
 import org.springframework.web.socket.WebSocketSession;
 import org.springframework.web.socket.client.WebSocketClient;
+import org.springframework.web.socket.handler.LoggingWebSocketHandlerDecorator;
 import org.springframework.web.socket.sockjs.transport.SockJsSession;
 import org.springframework.web.util.UriComponentsBuilder;
 
@@ -265,7 +266,9 @@ public class WebSocketStompClient extends StompClientSupport implements SmartLif
 		Assert.notNull(url, "'url' must not be null");
 		ConnectionHandlingStompSession session = createSession(connectHeaders, sessionHandler);
 		WebSocketTcpConnectionHandlerAdapter adapter = new WebSocketTcpConnectionHandlerAdapter(session);
-		getWebSocketClient().doHandshake(adapter, handshakeHeaders, url).addCallback(adapter);
+		getWebSocketClient()
+				.doHandshake(new LoggingWebSocketHandlerDecorator(adapter), handshakeHeaders, url)
+				.addCallback(adapter);
 		return session.getSessionFuture();
 	}
 

spring-websocket/src/test/java/org/springframework/web/socket/messaging/WebSocketStompClientTests.java

@@ -45,6 +45,7 @@ import org.springframework.web.socket.TextMessage;
 import org.springframework.web.socket.WebSocketHandler;
 import org.springframework.web.socket.WebSocketSession;
 import org.springframework.web.socket.client.WebSocketClient;
+import org.springframework.web.socket.handler.WebSocketHandlerDecorator;
 
 import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
@@ -317,9 +318,12 @@ public class WebSocketStompClientTests {
 
 	@SuppressWarnings("unchecked")
 	private TcpConnection<byte[]> getTcpConnection() throws Exception {
-		WebSocketHandler webSocketHandler = connect();
-		webSocketHandler.afterConnectionEstablished(this.webSocketSession);
-		return (TcpConnection<byte[]>) webSocketHandler;
+		WebSocketHandler handler = connect();
+		handler.afterConnectionEstablished(this.webSocketSession);
+		if (handler instanceof WebSocketHandlerDecorator) {
+			handler = ((WebSocketHandlerDecorator) handler).getLastHandler();
+		}
+		return (TcpConnection<byte[]>) handler;
 	}
 
 	private void testInactivityTaskScheduling(Runnable runnable, long delay, long sleepTime)

src/docs/asciidoc/web/webflux.adoc

@@ -761,6 +761,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>>#