Merge branch '6.1.x'

spring-web/src/main/java/org/springframework/web/server/adapter/DefaultServerWebExchange.java

@@ -24,7 +24,9 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.context.ApplicationContext;
@@ -249,13 +251,10 @@ public class DefaultServerWebExchange implements ServerWebExchange {
 	public Mono<Void> cleanupMultipart() {
 		return Mono.defer(() -> {
 			if (this.multipartRead) {
-				return getMultipartData()
+				return Mono.usingWhen(getMultipartData().onErrorComplete().map(this::collectParts),
-						.onErrorComplete()
+						parts -> Mono.empty(),
-						.flatMapIterable(Map::values)
+						parts -> Flux.fromIterable(parts).flatMap(part -> part.delete().onErrorComplete())
-						.flatMapIterable(Function.identity())
+				);
-						.flatMap(part -> part.delete()
-									.onErrorComplete())
-						.then();
 			}
 			else {
 				return Mono.empty();
@@ -263,6 +262,10 @@ public class DefaultServerWebExchange implements ServerWebExchange {
 		});
 	}
 
+	private List<Part> collectParts(MultiValueMap<String, Part> multipartData) {
+		return multipartData.values().stream().flatMap(List::stream).collect(Collectors.toList());
+	}
+
 	@Override
 	public LocaleContext getLocaleContext() {
 		return this.localeContextResolver.resolveLocaleContext(this);