Ensure multipart data is deleted in WebFlux when connection terminates

Before this change temporary files would not consistently be deleted
when the connection which uploads the multipart files closes naturally.

This change uses the usingWhen Reactor operator to ensure that the
termination of the connection doesn't prevent individual file parts
from being deleted due to a cancellation signal.

See gh-31217
Closes gh-32638

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()
-						.onErrorComplete()
-						.flatMapIterable(Map::values)
-						.flatMapIterable(Function.identity())
-						.flatMap(part -> part.delete()
-									.onErrorComplete())
-						.then();
+				return Mono.usingWhen(getMultipartData().onErrorComplete().map(this::collectParts),
+						parts -> Mono.empty(),
+						parts -> Flux.fromIterable(parts).flatMap(part -> part.delete().onErrorComplete())
+				);
 			}
 			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);