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

Author: simonbasle

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

@@ -24,7 +24,9 @@
 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,20 +251,21 @@ public Mono<MultiValueMap<String, Part>> getMultipartData() {
 	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();
 			}
 		});
 	}
 
+	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);