Add Flux<Part> ServerWebExchange.getParts()

This commit introduces ServerWebExchange.getParts(), as an alternative,
streaming way of accessing multipart data.

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

@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.context.ApplicationContext;
@@ -138,9 +139,23 @@ public interface ServerWebExchange {
 	 * cached so that this method is safe to call more than once.
 	 * <p><strong>Note:</strong>the {@linkplain Part#content() contents} of each
 	 * part is not cached, and can only be read once.
+	 * @see #getParts()
 	 */
 	Mono<MultiValueMap<String, Part>> getMultipartData();
 
+	/**
+	 * Return the parts of a multipart request if the Content-Type is
+	 * {@code "multipart/form-data"} or an empty flux otherwise.
+	 * <p><strong>Note:</strong> calling this method causes the request body to
+	 * be read and parsed in full and the resulting {@code Flux} is
+	 * cached so that this method is safe to call more than once.
+	 * <p><strong>Note:</strong>the {@linkplain Part#content() contents} of each
+	 * part is not cached, and can only be read once.
+	 * @since 5.2
+	 * @see #getMultipartData()
+	 */
+	Flux<Part> getParts();
+
 	/**
 	 * Return the {@link LocaleContext} using the configured
 	 * {@link org.springframework.web.server.i18n.LocaleContextResolver}.

spring-web/src/main/java/org/springframework/web/server/ServerWebExchangeDecorator.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.
@@ -20,6 +20,7 @@ import java.time.Instant;
 import java.util.Map;
 import java.util.function.Function;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.context.ApplicationContext;
@@ -107,6 +108,11 @@ public class ServerWebExchangeDecorator implements ServerWebExchange {
 		return getDelegate().getMultipartData();
 	}
 
+	@Override
+	public Flux<Part> getParts() {
+		return getDelegate().getParts();
+	}
+
 	@Override
 	public boolean isNotModified() {
 		return getDelegate().isNotModified();

spring-web/src/main/java/org/springframework/web/server/adapter/DefaultServerWebExchange.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,6 +25,7 @@ import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.context.ApplicationContext;
@@ -65,8 +66,7 @@ public class DefaultServerWebExchange implements ServerWebExchange {
 	private static final ResolvableType FORM_DATA_TYPE =
 			ResolvableType.forClassWithGenerics(MultiValueMap.class, String.class, String.class);
 
-	private static final ResolvableType MULTIPART_DATA_TYPE = ResolvableType.forClassWithGenerics(
+	private static final ResolvableType PARTS_DATA_TYPE = ResolvableType.forClass(Part.class);
-			MultiValueMap.class, String.class, Part.class);
 
 	private static final Mono<MultiValueMap<String, String>> EMPTY_FORM_DATA =
 			Mono.just(CollectionUtils.unmodifiableMultiValueMap(new LinkedMultiValueMap<String, String>(0)))
@@ -91,6 +91,8 @@ public class DefaultServerWebExchange implements ServerWebExchange {
 
 	private final Mono<MultiValueMap<String, Part>> multipartDataMono;
 
+	private final Flux<Part> partFlux;
+
 	@Nullable
 	private final ApplicationContext applicationContext;
 
@@ -129,7 +131,8 @@ public class DefaultServerWebExchange implements ServerWebExchange {
 		this.sessionMono = sessionManager.getSession(this).cache();
 		this.localeContextResolver = localeContextResolver;
 		this.formDataMono = initFormData(request, codecConfigurer, getLogPrefix());
-		this.multipartDataMono = initMultipartData(request, codecConfigurer, getLogPrefix());
+		this.partFlux = initParts(request, codecConfigurer, getLogPrefix());
+		this.multipartDataMono = initMultipartData(this.partFlux);
 		this.applicationContext = applicationContext;
 	}
 
@@ -156,27 +159,33 @@ public class DefaultServerWebExchange implements ServerWebExchange {
 	}
 
 	@SuppressWarnings("unchecked")
-	private static Mono<MultiValueMap<String, Part>> initMultipartData(ServerHttpRequest request,
+	private static Flux<Part> initParts(ServerHttpRequest request, ServerCodecConfigurer configurer, String logPrefix) {
-			ServerCodecConfigurer configurer, String logPrefix) {
-
 		try {
 			MediaType contentType = request.getHeaders().getContentType();
 			if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(contentType)) {
-				return ((HttpMessageReader<MultiValueMap<String, Part>>) configurer.getReaders().stream()
+				return ((HttpMessageReader<Part>)configurer.getReaders().stream()
-						.filter(reader -> reader.canRead(MULTIPART_DATA_TYPE, MediaType.MULTIPART_FORM_DATA))
+						.filter(reader -> reader.canRead(PARTS_DATA_TYPE, MediaType.MULTIPART_FORM_DATA))
 						.findFirst()
 						.orElseThrow(() -> new IllegalStateException("No multipart HttpMessageReader.")))
-						.readMono(MULTIPART_DATA_TYPE, request, Hints.from(Hints.LOG_PREFIX_HINT, logPrefix))
+						.read(PARTS_DATA_TYPE, request, Hints.from(Hints.LOG_PREFIX_HINT, logPrefix))
-						.switchIfEmpty(EMPTY_MULTIPART_DATA)
 						.cache();
 			}
 		}
 		catch (InvalidMediaTypeException ex) {
 			// Ignore
 		}
-		return EMPTY_MULTIPART_DATA;
+		return Flux.empty();
 	}
 
+	private static Mono<MultiValueMap<String, Part>> initMultipartData(Flux<Part> parts) {
+		return parts.collect(
+				() -> (MultiValueMap<String, Part>) new LinkedMultiValueMap<String, Part>(),
+				(map, part) -> map.add(part.name(), part))
+				.switchIfEmpty(EMPTY_MULTIPART_DATA)
+				.cache();
+	}
+
+
 
 	@Override
 	public ServerHttpRequest getRequest() {
@@ -221,6 +230,11 @@ public class DefaultServerWebExchange implements ServerWebExchange {
 		return this.multipartDataMono;
 	}
 
+	@Override
+	public Flux<Part> getParts() {
+		return this.partFlux;
+	}
+
 	@Override
 	public LocaleContext getLocaleContext() {
 		return this.localeContextResolver.resolveLocaleContext(this);