Polish during review of DataBuffer handling

spring-core/src/main/java/org/springframework/core/codec/AbstractDataBufferDecoder.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.
@@ -54,17 +54,17 @@ public abstract class AbstractDataBufferDecoder<T> extends AbstractDecoder<T> {
 
 
 	@Override
-	public Flux<T> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+	public Flux<T> decode(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
-		return Flux.from(inputStream).map(buffer -> decodeDataBuffer(buffer, elementType, mimeType, hints));
+		return Flux.from(input).map(buffer -> decodeDataBuffer(buffer, elementType, mimeType, hints));
 	}
 
 	@Override
-	public Mono<T> decodeToMono(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+	public Mono<T> decodeToMono(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
-		return DataBufferUtils.join(inputStream)
+		return DataBufferUtils.join(input)
 				.map(buffer -> decodeDataBuffer(buffer, elementType, mimeType, hints));
 	}
 

spring-core/src/main/java/org/springframework/core/codec/AbstractDecoder.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.
@@ -78,7 +78,12 @@ public abstract class AbstractDecoder<T> implements Decoder<T> {
 		if (mimeType == null) {
 			return true;
 		}
-		return this.decodableMimeTypes.stream().anyMatch(candidate -> candidate.isCompatibleWith(mimeType));
+		for (MimeType candidate : this.decodableMimeTypes) {
+			if (candidate.isCompatibleWith(mimeType)) {
+				return true;
+			}
+		}
+		return false;
 	}
 
 	@Override

spring-core/src/main/java/org/springframework/core/codec/AbstractEncoder.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.
@@ -74,7 +74,12 @@ public abstract class AbstractEncoder<T> implements Encoder<T> {
 		if (mimeType == null) {
 			return true;
 		}
-		return this.encodableMimeTypes.stream().anyMatch(candidate -> candidate.isCompatibleWith(mimeType));
+		for (MimeType candidate : this.encodableMimeTypes) {
+			if (candidate.isCompatibleWith(mimeType)) {
+				return true;
+			}
+		}
+		return false;
 	}
 
 }

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

@@ -57,10 +57,10 @@ public class DataBufferDecoder extends AbstractDataBufferDecoder<DataBuffer> {
 	}
 
 	@Override
-	public Flux<DataBuffer> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+	public Flux<DataBuffer> decode(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
-		return Flux.from(inputStream);
+		return Flux.from(input);
 	}
 
 	@Override

spring-core/src/main/java/org/springframework/core/codec/ResourceEncoder.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.
@@ -65,15 +65,14 @@ public class ResourceEncoder extends AbstractSingleValueEncoder<Resource> {
 	}
 
 	@Override
-	protected Flux<DataBuffer> encode(Resource resource, DataBufferFactory dataBufferFactory,
+	protected Flux<DataBuffer> encode(Resource resource, DataBufferFactory bufferFactory,
 			ResolvableType type, @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
 		if (logger.isDebugEnabled() && !Hints.isLoggingSuppressed(hints)) {
 			String logPrefix = Hints.getLogPrefix(hints);
 			logger.debug(logPrefix + "Writing [" + resource + "]");
 		}
-
-		return DataBufferUtils.read(resource, dataBufferFactory, this.bufferSize);
+		return DataBufferUtils.read(resource, bufferFactory, this.bufferSize);
 	}
 
 }

spring-core/src/main/java/org/springframework/core/codec/ResourceRegionEncoder.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,16 +76,16 @@ public class ResourceRegionEncoder extends AbstractEncoder<ResourceRegion> {
 	}
 
 	@Override
-	public Flux<DataBuffer> encode(Publisher<? extends ResourceRegion> inputStream,
+	public Flux<DataBuffer> encode(Publisher<? extends ResourceRegion> input,
 			DataBufferFactory bufferFactory, ResolvableType elementType, @Nullable MimeType mimeType,
 			@Nullable Map<String, Object> hints) {
 
-		Assert.notNull(inputStream, "'inputStream' must not be null");
+		Assert.notNull(input, "'inputStream' must not be null");
 		Assert.notNull(bufferFactory, "'bufferFactory' must not be null");
 		Assert.notNull(elementType, "'elementType' must not be null");
 
-		if (inputStream instanceof Mono) {
-			return Mono.from(inputStream)
+		if (input instanceof Mono) {
+			return Mono.from(input)
 					.flatMapMany(region -> {
 						if (!region.getResource().isReadable()) {
 							return Flux.error(new EncodingException(
@@ -96,32 +96,25 @@ public class ResourceRegionEncoder extends AbstractEncoder<ResourceRegion> {
 		}
 		else {
 			final String boundaryString = Hints.getRequiredHint(hints, BOUNDARY_STRING_HINT);
-			byte[] startBoundary = getAsciiBytes("\r\n--" + boundaryString + "\r\n");
-			byte[] contentType = mimeType != null ? getAsciiBytes("Content-Type: " + mimeType + "\r\n") : new byte[0];
+			byte[] startBoundary = toAsciiBytes("\r\n--" + boundaryString + "\r\n");
+			byte[] contentType = mimeType != null ? toAsciiBytes("Content-Type: " + mimeType + "\r\n") : new byte[0];
 
-			return Flux.from(inputStream).
-					concatMap(region -> {
+			return Flux.from(input)
+					.concatMap(region -> {
 						if (!region.getResource().isReadable()) {
 							return Flux.error(new EncodingException(
 									"Resource " + region.getResource() + " is not readable"));
 						}
-						else {
-							return Flux.concat(
-									getRegionPrefix(bufferFactory, startBoundary, contentType, region),
-									writeResourceRegion(region, bufferFactory, hints));
-						}
+						Flux<DataBuffer> prefix = Flux.just(
+								bufferFactory.wrap(startBoundary),
+								bufferFactory.wrap(contentType),
+								bufferFactory.wrap(getContentRangeHeader(region))); // only wrapping, no allocation
+
+						return prefix.concatWith(writeResourceRegion(region, bufferFactory, hints));
 					})
-					.concatWith(getRegionSuffix(bufferFactory, boundaryString));
+					.concatWithValues(getRegionSuffix(bufferFactory, boundaryString));
 		}
-	}
-
-	private Flux<DataBuffer> getRegionPrefix(DataBufferFactory bufferFactory, byte[] startBoundary,
-			byte[] contentType, ResourceRegion region) {
-
-		return Flux.just(
-				bufferFactory.wrap(startBoundary),
-				bufferFactory.wrap(contentType),
-				bufferFactory.wrap(getContentRangeHeader(region))); // only wrapping, no allocation
+		// No doOnDiscard (no caching after DataBufferUtils#read)
 	}
 
 	private Flux<DataBuffer> writeResourceRegion(
@@ -140,12 +133,12 @@ public class ResourceRegionEncoder extends AbstractEncoder<ResourceRegion> {
 		return DataBufferUtils.takeUntilByteCount(in, count);
 	}
 
-	private Flux<DataBuffer> getRegionSuffix(DataBufferFactory bufferFactory, String boundaryString) {
-		byte[] endBoundary = getAsciiBytes("\r\n--" + boundaryString + "--");
-		return Flux.just(bufferFactory.wrap(endBoundary));
+	private DataBuffer getRegionSuffix(DataBufferFactory bufferFactory, String boundaryString) {
+		byte[] endBoundary = toAsciiBytes("\r\n--" + boundaryString + "--");
+		return bufferFactory.wrap(endBoundary);
 	}
 
-	private byte[] getAsciiBytes(String in) {
+	private byte[] toAsciiBytes(String in) {
 		return in.getBytes(StandardCharsets.US_ASCII);
 	}
 
@@ -155,10 +148,10 @@ public class ResourceRegionEncoder extends AbstractEncoder<ResourceRegion> {
 		OptionalLong contentLength = contentLength(region.getResource());
 		if (contentLength.isPresent()) {
 			long length = contentLength.getAsLong();
-			return getAsciiBytes("Content-Range: bytes " + start + '-' + end + '/' + length + "\r\n\r\n");
+			return toAsciiBytes("Content-Range: bytes " + start + '-' + end + '/' + length + "\r\n\r\n");
 		}
 		else {
-			return getAsciiBytes("Content-Range: bytes " + start + '-' + end + "\r\n\r\n");
+			return toAsciiBytes("Content-Range: bytes " + start + '-' + end + "\r\n\r\n");
 		}
 	}
 

spring-core/src/main/java/org/springframework/core/codec/StringDecoder.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 java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -88,14 +87,14 @@ public final class StringDecoder extends AbstractDataBufferDecoder<String> {
 	}
 
 	@Override
-	public Flux<String> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+	public Flux<String> decode(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
 		List<byte[]> delimiterBytes = getDelimiterBytes(mimeType);
 
-		Flux<DataBuffer> inputFlux = Flux.from(inputStream)
-				.flatMapIterable(dataBuffer -> splitOnDelimiter(dataBuffer, delimiterBytes))
-				.bufferUntil(StringDecoder::isEndFrame)
+		Flux<DataBuffer> inputFlux = Flux.from(input)
+				.flatMapIterable(buffer -> splitOnDelimiter(buffer, delimiterBytes))
+				.bufferUntil(buffer -> buffer == END_FRAME)
 				.map(StringDecoder::joinUntilEndFrame)
 				.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release);
 
@@ -103,51 +102,60 @@ public final class StringDecoder extends AbstractDataBufferDecoder<String> {
 	}
 
 	private List<byte[]> getDelimiterBytes(@Nullable MimeType mimeType) {
-		return this.delimitersCache.computeIfAbsent(getCharset(mimeType),
-				charset -> this.delimiters.stream()
-						.map(s -> s.getBytes(charset))
-						.collect(Collectors.toList()));
+		return this.delimitersCache.computeIfAbsent(getCharset(mimeType), charset -> {
+			List<byte[]> list = new ArrayList<>();
+			for (String delimiter : this.delimiters) {
+				byte[] bytes = delimiter.getBytes(charset);
+				list.add(bytes);
+			}
+			return list;
+		});
 	}
 
 	/**
 	 * Split the given data buffer on delimiter boundaries.
 	 * The returned Flux contains an {@link #END_FRAME} buffer after each delimiter.
 	 */
-	private List<DataBuffer> splitOnDelimiter(DataBuffer dataBuffer, List<byte[]> delimiterBytes) {
+	private List<DataBuffer> splitOnDelimiter(DataBuffer buffer, List<byte[]> delimiterBytes) {
 		List<DataBuffer> frames = new ArrayList<>();
-		do {
-			int length = Integer.MAX_VALUE;
-			byte[] matchingDelimiter = null;
-			for (byte[] delimiter : delimiterBytes) {
-				int index = indexOf(dataBuffer, delimiter);
-				if (index >= 0 && index < length) {
-					length = index;
-					matchingDelimiter = delimiter;
+		try {
+			do {
+				int length = Integer.MAX_VALUE;
+				byte[] matchingDelimiter = null;
+				for (byte[] delimiter : delimiterBytes) {
+					int index = indexOf(buffer, delimiter);
+					if (index >= 0 && index < length) {
+						length = index;
+						matchingDelimiter = delimiter;
+					}
 				}
-			}
-			DataBuffer frame;
-			int readPosition = dataBuffer.readPosition();
-			if (matchingDelimiter != null) {
-				if (this.stripDelimiter) {
-					frame = dataBuffer.slice(readPosition, length);
+				DataBuffer frame;
+				int readPosition = buffer.readPosition();
+				if (matchingDelimiter != null) {
+					frame = this.stripDelimiter ?
+							buffer.slice(readPosition, length) :
+							buffer.slice(readPosition, length + matchingDelimiter.length);
+					buffer.readPosition(readPosition + length + matchingDelimiter.length);
+					frames.add(DataBufferUtils.retain(frame));
+					frames.add(END_FRAME);
 				}
 				else {
-					frame = dataBuffer.slice(readPosition, length + matchingDelimiter.length);
+					frame = buffer.slice(readPosition, buffer.readableByteCount());
+					buffer.readPosition(readPosition + buffer.readableByteCount());
+					frames.add(DataBufferUtils.retain(frame));
 				}
-				dataBuffer.readPosition(readPosition + length + matchingDelimiter.length);
-
-				frames.add(DataBufferUtils.retain(frame));
-				frames.add(END_FRAME);
 			}
-			else {
-				frame = dataBuffer.slice(readPosition, dataBuffer.readableByteCount());
-				dataBuffer.readPosition(readPosition + dataBuffer.readableByteCount());
-				frames.add(DataBufferUtils.retain(frame));
+			while (buffer.readableByteCount() > 0);
+		}
+		catch (Throwable ex) {
+			for (DataBuffer frame : frames) {
+				DataBufferUtils.release(frame);
 			}
+			throw ex;
+		}
+		finally {
+			DataBufferUtils.release(buffer);
 		}
-		while (dataBuffer.readableByteCount() > 0);
-
-		DataBufferUtils.release(dataBuffer);
 		return frames;
 	}
 
@@ -155,44 +163,38 @@ public final class StringDecoder extends AbstractDataBufferDecoder<String> {
 	 * Find the given delimiter in the given data buffer.
 	 * @return the index of the delimiter, or -1 if not found.
 	 */
-	private static int indexOf(DataBuffer dataBuffer, byte[] delimiter) {
-		for (int i = dataBuffer.readPosition(); i < dataBuffer.writePosition(); i++) {
-			int dataBufferPos = i;
+	private static int indexOf(DataBuffer buffer, byte[] delimiter) {
+		for (int i = buffer.readPosition(); i < buffer.writePosition(); i++) {
+			int bufferPos = i;
 			int delimiterPos = 0;
 			while (delimiterPos < delimiter.length) {
-				if (dataBuffer.getByte(dataBufferPos) != delimiter[delimiterPos]) {
+				if (buffer.getByte(bufferPos) != delimiter[delimiterPos]) {
 					break;
 				}
 				else {
-					dataBufferPos++;
-					if (dataBufferPos == dataBuffer.writePosition() &&
-							delimiterPos != delimiter.length - 1) {
+					bufferPos++;
+					boolean endOfBuffer = bufferPos == buffer.writePosition();
+					boolean endOfDelimiter = delimiterPos == delimiter.length - 1;
+					if (endOfBuffer && !endOfDelimiter) {
 						return -1;
 					}
 				}
 				delimiterPos++;
 			}
 			if (delimiterPos == delimiter.length) {
-				return i - dataBuffer.readPosition();
+				return i - buffer.readPosition();
 			}
 		}
 		return -1;
 	}
 
-	/**
-	 * Check whether the given buffer is {@link #END_FRAME}.
-	 */
-	private static boolean isEndFrame(DataBuffer dataBuffer) {
-		return dataBuffer == END_FRAME;
-	}
-
 	/**
 	 * Join the given list of buffers into a single buffer.
 	 */
 	private static DataBuffer joinUntilEndFrame(List<DataBuffer> dataBuffers) {
 		if (!dataBuffers.isEmpty()) {
 			int lastIdx = dataBuffers.size() - 1;
-			if (isEndFrame(dataBuffers.get(lastIdx))) {
+			if (dataBuffers.get(lastIdx) == END_FRAME) {
 				dataBuffers.remove(lastIdx);
 			}
 		}

spring-core/src/test/java/org/springframework/core/codec/ResourceRegionEncoderTests.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,6 +19,8 @@ package org.springframework.core.codec;
 import java.util.Collections;
 import java.util.function.Consumer;
 
+import io.netty.buffer.PooledByteBufAllocator;
+import org.junit.After;
 import org.junit.Test;
 import org.reactivestreams.Subscription;
 import reactor.core.publisher.BaseSubscriber;
@@ -32,6 +34,7 @@ import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.LeakAwareDataBufferFactory;
+import org.springframework.core.io.buffer.NettyDataBufferFactory;
 import org.springframework.core.io.buffer.support.DataBufferTestUtils;
 import org.springframework.core.io.support.ResourceRegion;
 import org.springframework.util.MimeType;
@@ -48,9 +51,15 @@ public class ResourceRegionEncoderTests  {
 
 	private ResourceRegionEncoder encoder = new ResourceRegionEncoder();
 
-	private LeakAwareDataBufferFactory bufferFactory = new LeakAwareDataBufferFactory();
+	private LeakAwareDataBufferFactory bufferFactory =
+			new LeakAwareDataBufferFactory(new NettyDataBufferFactory(PooledByteBufAllocator.DEFAULT));
 
 
+	@After
+	public void tearDown() throws Exception {
+		this.bufferFactory.checkForLeaks();
+	}
+
 	@Test
 	public void canEncode() {
 		ResolvableType resourceRegion = ResolvableType.forClass(ResourceRegion.class);
@@ -79,8 +88,6 @@ public class ResourceRegionEncoderTests  {
 				.consumeNextWith(stringConsumer("Spring"))
 				.expectComplete()
 				.verify();
-
-		 this.bufferFactory.checkForLeaks();
 	}
 
 	@Test
@@ -120,8 +127,6 @@ public class ResourceRegionEncoderTests  {
 				.consumeNextWith(stringConsumer("\r\n--" + boundary + "--"))
 				.expectComplete()
 				.verify();
-
-		 this.bufferFactory.checkForLeaks();
 	}
 
 	@Test // gh-22107
@@ -144,8 +149,23 @@ public class ResourceRegionEncoderTests  {
 		ZeroDemandSubscriber subscriber = new ZeroDemandSubscriber();
 		flux.subscribe(subscriber);
 		subscriber.cancel();
+	}
 
-		this.bufferFactory.checkForLeaks();
+	@Test // gh-22107
+	public void cancelWithoutDemandForSingleResourceRegion() {
+		Resource resource = new ClassPathResource("ResourceRegionEncoderTests.txt", getClass());
+		Mono<ResourceRegion> regions = Mono.just(new ResourceRegion(resource, 0, 6));
+		String boundary = MimeTypeUtils.generateMultipartBoundaryString();
+
+		Flux<DataBuffer> flux = this.encoder.encode(regions, this.bufferFactory,
+				ResolvableType.forClass(ResourceRegion.class),
+				MimeType.valueOf("text/plain"),
+				Collections.singletonMap(ResourceRegionEncoder.BOUNDARY_STRING_HINT, boundary)
+		);
+
+		ZeroDemandSubscriber subscriber = new ZeroDemandSubscriber();
+		flux.subscribe(subscriber);
+		subscriber.cancel();
 	}
 
 	@Test
@@ -170,14 +190,11 @@ public class ResourceRegionEncoderTests  {
 				.consumeNextWith(stringConsumer("Spring"))
 				.expectError(EncodingException.class)
 				.verify();
-
-		 this.bufferFactory.checkForLeaks();
 	}
 
 	protected Consumer<DataBuffer> stringConsumer(String expected) {
 		return dataBuffer -> {
-			String value =
-					DataBufferTestUtils.dumpString(dataBuffer, UTF_8);
+			String value = DataBufferTestUtils.dumpString(dataBuffer, UTF_8);
 			DataBufferUtils.release(dataBuffer);
 			assertEquals(expected, value);
 		};

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

@@ -125,8 +125,9 @@ public class EncoderHttpMessageWriter<T> implements HttpMessageWriter<T> {
 					}))
 					.flatMap(buffer -> {
 						headers.setContentLength(buffer.readableByteCount());
-						return message.writeWith(Mono.fromCallable(() -> buffer)
-								.doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release));
+						return message.writeWith(
+								Mono.fromCallable(() -> buffer)
+										.doOnDiscard(PooledDataBuffer.class, PooledDataBuffer::release));
 					});
 		}
 
@@ -162,10 +163,16 @@ public class EncoderHttpMessageWriter<T> implements HttpMessageWriter<T> {
 	}
 
 	private boolean isStreamingMediaType(@Nullable MediaType contentType) {
-		return (contentType != null && this.encoder instanceof HttpMessageEncoder &&
-				((HttpMessageEncoder<?>) this.encoder).getStreamingMediaTypes().stream()
-						.anyMatch(streamingMediaType -> contentType.isCompatibleWith(streamingMediaType) &&
-								contentType.getParameters().entrySet().containsAll(streamingMediaType.getParameters().keySet())));
+		if (contentType == null || !(this.encoder instanceof HttpMessageEncoder)) {
+			return false;
+		}
+		for (MediaType mediaType : ((HttpMessageEncoder<?>) this.encoder).getStreamingMediaTypes()) {
+			if (contentType.isCompatibleWith(mediaType) &&
+					contentType.getParameters().entrySet().containsAll(mediaType.getParameters().keySet())) {
+				return true;
+			}
+		}
+		return false;
 	}
 
 

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

@@ -56,7 +56,7 @@ public class FormHttpMessageReader extends LoggingCodecSupport
 	 */
 	public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
 
-	private static final ResolvableType MULTIVALUE_TYPE =
+	private static final ResolvableType MULTIVALUE_STRINGS_TYPE =
 			ResolvableType.forClassWithGenerics(MultiValueMap.class, String.class, String.class);
 
 
@@ -83,9 +83,11 @@ public class FormHttpMessageReader extends LoggingCodecSupport
 
 	@Override
 	public boolean canRead(ResolvableType elementType, @Nullable MediaType mediaType) {
-		return ((MULTIVALUE_TYPE.isAssignableFrom(elementType) ||
-				(elementType.hasUnresolvableGenerics() &&
-						MultiValueMap.class.isAssignableFrom(elementType.toClass()))) &&
+		boolean multiValueUnresolved =
+				elementType.hasUnresolvableGenerics() &&
+						MultiValueMap.class.isAssignableFrom(elementType.toClass());
+
+		return ((MULTIVALUE_STRINGS_TYPE.isAssignableFrom(elementType) || multiValueUnresolved) &&
 				(mediaType == null || MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)));
 	}
 

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

@@ -164,8 +164,8 @@ public class ServerSentEventHttpMessageReader implements HttpMessageReader<Objec
 		}
 
 		byte[] bytes = data.getBytes(StandardCharsets.UTF_8);
-		Mono<DataBuffer> input = Mono.just(bufferFactory.wrap(bytes));
-		return this.decoder.decodeToMono(input, dataType, MediaType.TEXT_EVENT_STREAM, hints);
+		DataBuffer buffer = bufferFactory.wrap(bytes);  // wrapping only, no allocation
+		return this.decoder.decodeToMono(Mono.just(buffer), dataType, MediaType.TEXT_EVENT_STREAM, hints);
 	}
 
 	@Override

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

@@ -184,7 +184,7 @@ public class ServerSentEventHttpMessageWriter implements HttpMessageWriter<Objec
 	private Mono<DataBuffer> encodeText(CharSequence text, MediaType mediaType, DataBufferFactory bufferFactory) {
 		Assert.notNull(mediaType.getCharset(), "Expected MediaType with charset");
 		byte[] bytes = text.toString().getBytes(mediaType.getCharset());
-		return Mono.fromCallable(() -> bufferFactory.wrap(bytes)); // wrapping, not allocating
+		return Mono.just(bufferFactory.wrap(bytes)); // wrapping, not allocating
 	}
 
 	@Override

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

@@ -111,13 +111,13 @@ public class Jaxb2XmlEncoder extends AbstractSingleValueEncoder<Object> {
 		return Flux.defer(() -> {
 			boolean release = true;
 			DataBuffer buffer = bufferFactory.allocateBuffer(1024);
-			OutputStream outputStream = buffer.asOutputStream();
-			Class<?> clazz = ClassUtils.getUserClass(value);
 			try {
+				OutputStream outputStream = buffer.asOutputStream();
+				Class<?> clazz = ClassUtils.getUserClass(value);
 				Marshaller marshaller = initMarshaller(clazz);
 				marshaller.marshal(value, outputStream);
 				release = false;
-				return Mono.fromCallable(() -> buffer);  // Rely on doOnDiscard in base class
+				return Mono.fromCallable(() -> buffer);  // relying on doOnDiscard in base class
 			}
 			catch (MarshalException ex) {
 				return Flux.error(new EncodingException(

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

@@ -33,8 +33,8 @@ import com.fasterxml.aalto.AsyncXMLStreamReader;
 import com.fasterxml.aalto.evt.EventAllocatorImpl;
 import com.fasterxml.aalto.stax.InputFactoryImpl;
 import org.reactivestreams.Publisher;
+import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.AbstractDecoder;
@@ -95,27 +95,30 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
 
 	@Override
 	@SuppressWarnings({"rawtypes", "unchecked"})  // on JDK 9 where XMLEventReader is Iterator<Object>
-	public Flux<XMLEvent> decode(Publisher<DataBuffer> inputStream, ResolvableType elementType,
+	public Flux<XMLEvent> decode(Publisher<DataBuffer> input, ResolvableType elementType,
 			@Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
-		Flux<DataBuffer> flux = Flux.from(inputStream);
 		if (this.useAalto) {
-			AaltoDataBufferToXmlEvent aaltoMapper = new AaltoDataBufferToXmlEvent();
-			return flux.flatMap(aaltoMapper)
-					.doFinally(signalType -> aaltoMapper.endOfInput());
+			AaltoDataBufferToXmlEvent mapper = new AaltoDataBufferToXmlEvent();
+			return Flux.from(input)
+					.flatMapIterable(mapper)
+					.doFinally(signalType -> mapper.endOfInput());
 		}
 		else {
-			Mono<DataBuffer> singleBuffer = DataBufferUtils.join(flux);
-			return singleBuffer.
-					flatMapMany(dataBuffer -> {
+			return DataBufferUtils.join(input).
+					flatMapIterable(buffer -> {
 						try {
-							InputStream is = dataBuffer.asInputStream();
+							InputStream is = buffer.asInputStream();
 							Iterator eventReader = inputFactory.createXMLEventReader(is);
-							return Flux.fromIterable((Iterable<XMLEvent>) () -> eventReader)
-									.doFinally(t -> DataBufferUtils.release(dataBuffer));
+							List<XMLEvent> result = new ArrayList<>();
+							eventReader.forEachRemaining(event -> result.add((XMLEvent) event));
+							return result;
 						}
 						catch (XMLStreamException ex) {
-							return Mono.error(ex);
+							throw Exceptions.propagate(ex);
+						}
+						finally {
+							DataBufferUtils.release(buffer);
 						}
 					});
 		}
@@ -125,7 +128,7 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
 	/*
 	 * Separate static class to isolate Aalto dependency.
 	 */
-	private static class AaltoDataBufferToXmlEvent implements Function<DataBuffer, Publisher<? extends XMLEvent>> {
+	private static class AaltoDataBufferToXmlEvent implements Function<DataBuffer, List<? extends XMLEvent>> {
 
 		private static final AsyncXMLInputFactory inputFactory =
 				StaxUtils.createDefensiveInputFactory(InputFactoryImpl::new);
@@ -137,7 +140,7 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
 
 
 		@Override
-		public Publisher<? extends XMLEvent> apply(DataBuffer dataBuffer) {
+		public List<? extends XMLEvent> apply(DataBuffer dataBuffer) {
 			try {
 				this.streamReader.getInputFeeder().feedInput(dataBuffer.asByteBuffer());
 				List<XMLEvent> events = new ArrayList<>();
@@ -154,10 +157,10 @@ public class XmlEventDecoder extends AbstractDecoder<XMLEvent> {
 						}
 					}
 				}
-				return Flux.fromIterable(events);
+				return events;
 			}
 			catch (XMLStreamException ex) {
-				return Mono.error(ex);
+				throw Exceptions.propagate(ex);
 			}
 			finally {
 				DataBufferUtils.release(dataBuffer);

spring-web/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java

@@ -181,8 +181,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse {
 
 	@Override
 	public final Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
-		return new ChannelSendOperator<>(body,
-				writePublisher -> doCommit(() -> writeAndFlushWithInternal(writePublisher)))
+		return new ChannelSendOperator<>(body, inner -> doCommit(() -> writeAndFlushWithInternal(inner)))
 				.doOnError(t -> removeContentLength());
 	}