Support Publishers for multipart data in BodyInserters

This commit uses the changes in the previous commit to support
Publishers as parts for multipart data.

Issue: SPR-16307

Author: poutsma

spring-webflux/src/main/java/org/springframework/web/reactive/function/BodyInserters.java

@@ -17,6 +17,7 @@
 package org.springframework.web.reactive.function;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
@@ -27,8 +28,10 @@
 import org.springframework.core.ResolvableType;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.http.HttpEntity;
 import org.springframework.http.MediaType;
 import org.springframework.http.ReactiveHttpOutputMessage;
+import org.springframework.http.client.MultipartBodyBuilder;
 import org.springframework.http.client.reactive.ClientHttpRequest;
 import org.springframework.http.codec.HttpMessageWriter;
 import org.springframework.http.codec.ServerSentEvent;
@@ -204,14 +207,11 @@ public static <T, S extends Publisher<ServerSentEvent<T>>> BodyInserter<S, Serve
 	 * @param formData the form data to write to the output message
 	 * @return a {@code FormInserter} that writes form data
 	 */
-	// Note that the returned FormInserter is parameterized to ClientHttpRequest, not
-	// ReactiveHttpOutputMessage like other methods, since sending form data only typically happens
-	// on the client-side
 	public static FormInserter<String> fromFormData(MultiValueMap<String, String> formData) {
 
 		Assert.notNull(formData, "'formData' must not be null");
 
-		return DefaultFormInserter.forFormData().with(formData);
+		return new DefaultFormInserter().with(formData);
 	}
 
 	/**
@@ -222,14 +222,11 @@ public static FormInserter<String> fromFormData(MultiValueMap<String, String> fo
 	 * @param value the value to add to the form
 	 * @return a {@code FormInserter} that writes form data
 	 */
-	// Note that the returned FormInserter is parameterized to ClientHttpRequest, not
-	// ReactiveHttpOutputMessage like other methods, since sending form data only typically happens
-	// on the client-side
 	public static FormInserter<String> fromFormData(String key, String value) {
 		Assert.notNull(key, "'key' must not be null");
 		Assert.notNull(value, "'value' must not be null");
 
-		return DefaultFormInserter.forFormData().with(key, value);
+		return new DefaultFormInserter().with(key, value);
 	}
 
 	/**
@@ -251,15 +248,11 @@ public static FormInserter<String> fromFormData(String key, String value) {
 	 *
 	 * @param multipartData the form data to write to the output message
 	 * @return a {@code BodyInserter} that writes multipart data
+	 * @see MultipartBodyBuilder
 	 */
-	// Note that the returned BodyInserter is parameterized to ClientHttpRequest, not
-	// ReactiveHttpOutputMessage like other methods, since sending form data only typically happens
-	// on the client-side
-	public static <T> FormInserter<T> fromMultipartData(MultiValueMap<String, T> multipartData) {
-
+	public static MultipartInserter fromMultipartData(MultiValueMap<String, Object> multipartData) {
 		Assert.notNull(multipartData, "'multipartData' must not be null");
-
-		return DefaultFormInserter.<T>forMultipartData().with(multipartData);
+		return new DefaultMultipartInserter().with(multipartData);
 	}
 
 	/**
@@ -271,14 +264,49 @@ public static <T> FormInserter<T> fromMultipartData(MultiValueMap<String, T> mul
 	 * @return a {@code FormInserter} that can writes the provided multipart
 	 * data and also allows adding more parts
 	 */
-	// Note that the returned BodyInserter is parameterized to ClientHttpRequest, not
-	// ReactiveHttpOutputMessage like other methods, since sending form data only typically happens
-	// on the client-side
-	public static <T> FormInserter<T> fromMultipartData(String key, T value) {
+	public static MultipartInserter fromMultipartData(String key, Object value) {
 		Assert.notNull(key, "'key' must not be null");
 		Assert.notNull(value, "'value' must not be null");
 
-		return DefaultFormInserter.<T>forMultipartData().with(key, value);
+		return new DefaultMultipartInserter().with(key, value);
+	}
+
+	/**
+	 * A variant of {@link #fromMultipartData(MultiValueMap)} for adding asynchronous data as a
+	 * part in-line vs building a {@code MultiValueMap} and passing it in.
+	 * @param key the part name
+	 * @param publisher the publisher that forms the part value
+	 * @param elementClass the class contained in the {@code publisher}
+	 * @return a {@code FormInserter} that can writes the provided multipart
+	 * data and also allows adding more parts
+	 */
+	public static <T, P extends Publisher<T>> MultipartInserter fromMultipartAsyncData(String key,
+			P publisher, Class<T> elementClass) {
+
+		Assert.notNull(key, "'key' must not be null");
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(elementClass, "'elementClass' must not be null");
+
+		return new DefaultMultipartInserter().withPublisher(key, publisher, elementClass);
+	}
+
+	/**
+	 * A variant of {@link #fromMultipartData(MultiValueMap)} for adding asynchronous data as a
+	 * part in-line vs building a {@code MultiValueMap} and passing it in.
+	 * @param key the part name
+	 * @param publisher the publisher that forms the part value
+	 * @param typeReference the type contained in the {@code publisher}
+	 * @return a {@code FormInserter} that can writes the provided multipart
+	 * data and also allows adding more parts
+	 */
+	public static <T, P extends Publisher<T>> MultipartInserter fromMultipartAsyncData(String key,
+			P publisher, ParameterizedTypeReference<T> typeReference) {
+
+		Assert.notNull(key, "'key' must not be null");
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(typeReference, "'typeReference' must not be null");
+
+		return new DefaultMultipartInserter().withPublisher(key, publisher, typeReference);
 	}
 
 	/**
@@ -350,6 +378,8 @@ private static <T> HttpMessageWriter<T> cast(HttpMessageWriter<?> messageWriter)
 	 * Sub-interface of {@link BodyInserter} that allows for additional (multipart) form data to be
 	 * added.
 	 */
+	// Note that FormInserter is parameterized to ClientHttpRequest, not ReactiveHttpOutputMessage
+	// like other return values methods, since sending form data only typically happens on the client-side
 	public interface FormInserter<T> extends
 			BodyInserter<MultiValueMap<String, T>, ClientHttpRequest> {
 
@@ -370,45 +400,113 @@ public interface FormInserter<T> extends
 
 	}
 
-	private static class DefaultFormInserter<T> implements FormInserter<T> {
 
-		private final MultiValueMap<String, T> data = new LinkedMultiValueMap<>();
+	/**
+	 * Extension of {@link FormInserter} that has methods for adding asynchronous part data.
+	 */
+	public interface MultipartInserter extends FormInserter<Object> {
+
+		/**
+		 * Adds the specified publisher as a part.
+		 *
+		 * @param key the key to be added
+		 * @param publisher the publisher to be added as value
+		 * @param elementClass the class of elements contained in {@code publisher}
+		 * @return this inserter
+		 */
+		<T, P extends Publisher<T>> MultipartInserter withPublisher(String key, P publisher,
+				Class<T> elementClass);
+
+		/**
+		 * Adds the specified publisher as a part.
+		 *
+		 * @param key the key to be added
+		 * @param publisher the publisher to be added as value
+		 * @param typeReference the type of elements contained in {@code publisher}
+		 * @return this inserter
+		 */
+		<T, P extends Publisher<T>> MultipartInserter withPublisher(String key, P publisher,
+				ParameterizedTypeReference<T> typeReference);
+
+	}
+
+
+	private static class DefaultFormInserter implements FormInserter<String> {
+
+		private final MultiValueMap<String, String> data = new LinkedMultiValueMap<>();
+
+		public DefaultFormInserter() {
+		}
+
+		@Override
+		public FormInserter<String> with(String key, @Nullable String value) {
+			this.data.add(key, value);
+			return this;
+		}
+
+		@Override
+		public FormInserter<String> with(MultiValueMap<String, String> values) {
+			this.data.addAll(values);
+			return this;
+		}
+
+		@Override
+		public Mono<Void> insert(ClientHttpRequest outputMessage, Context context) {
+			HttpMessageWriter<MultiValueMap<String, String>> messageWriter =
+					findMessageWriter(context, FORM_TYPE, MediaType.APPLICATION_FORM_URLENCODED);
+			return messageWriter.write(Mono.just(this.data), FORM_TYPE,
+					MediaType.APPLICATION_FORM_URLENCODED,
+					outputMessage, context.hints());
+		}
+	}
 
-		private final ResolvableType type;
 
-		private final MediaType mediaType;
+	private static class DefaultMultipartInserter implements MultipartInserter {
 
+		private final MultipartBodyBuilder builder = new MultipartBodyBuilder();
 
-		private DefaultFormInserter(ResolvableType type, MediaType mediaType) {
-			this.type = type;
-			this.mediaType = mediaType;
+		public DefaultMultipartInserter() {
 		}
 
-		public static FormInserter<String> forFormData() {
-			return new DefaultFormInserter<>(FORM_TYPE, MediaType.APPLICATION_FORM_URLENCODED);
+		@Override
+		public MultipartInserter with(String key, @Nullable Object value) {
+			Assert.notNull(value, "'value' must not be null");
+			this.builder.part(key, value);
+			return this;
 		}
 
-		public static <T> FormInserter<T> forMultipartData() {
-			return new DefaultFormInserter<>(MULTIPART_VALUE_TYPE, MediaType.MULTIPART_FORM_DATA);
+		@Override
+		public MultipartInserter with(MultiValueMap<String, Object> values) {
+			Assert.notNull(values, "'values' must not be null");
+			for (Map.Entry<String, List<Object>> entry : values.entrySet()) {
+				this.builder.part(entry.getKey(), entry.getValue());
+			}
+			return this;
 		}
 
 		@Override
-		public FormInserter<T> with(String key, @Nullable T value) {
-			this.data.add(key, value);
+		public <T, P extends Publisher<T>> MultipartInserter withPublisher(String key,
+				P publisher, Class<T> elementClass) {
+
+			this.builder.asyncPart(key, publisher, elementClass);
 			return this;
 		}
 
 		@Override
-		public FormInserter<T> with(MultiValueMap<String, T> values) {
-			this.data.addAll(values);
+		public <T, P extends Publisher<T>> MultipartInserter withPublisher(String key,
+				P publisher, ParameterizedTypeReference<T> typeReference) {
+
+			this.builder.asyncPart(key, publisher, typeReference);
 			return this;
 		}
 
 		@Override
 		public Mono<Void> insert(ClientHttpRequest outputMessage, Context context) {
-			HttpMessageWriter<MultiValueMap<String, T>> messageWriter =
-					findMessageWriter(context, this.type, this.mediaType);
-			return messageWriter.write(Mono.just(this.data), this.type, this.mediaType,
+			HttpMessageWriter<MultiValueMap<String, HttpEntity<?>>> messageWriter =
+					findMessageWriter(context, MULTIPART_VALUE_TYPE, MediaType.MULTIPART_FORM_DATA);
+			MultiValueMap<String, HttpEntity<?>> body = this.builder.build();
+			return messageWriter.write(Mono.just(body), MULTIPART_VALUE_TYPE,
+					MediaType.MULTIPART_FORM_DATA,
 					outputMessage, context.hints());
 		}
 	}

spring-webflux/src/test/java/org/springframework/web/reactive/function/BodyInsertersTests.java

@@ -53,6 +53,7 @@
 import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.http.codec.ServerSentEventHttpMessageWriter;
 import org.springframework.http.codec.json.Jackson2JsonEncoder;
+import org.springframework.http.codec.multipart.MultipartHttpMessageWriter;
 import org.springframework.http.codec.xml.Jaxb2XmlEncoder;
 import org.springframework.http.server.reactive.ServerHttpRequest;
 import org.springframework.http.server.reactive.ServerHttpResponse;
@@ -89,6 +90,7 @@ public void createContext() {
 		messageWriters.add(new ServerSentEventHttpMessageWriter(jsonEncoder));
 		messageWriters.add(new FormHttpMessageWriter());
 		messageWriters.add(new EncoderHttpMessageWriter<>(CharSequenceEncoder.allMimeTypes()));
+		messageWriters.add(new MultipartHttpMessageWriter(messageWriters));
 
 		this.context = new BodyInserter.Context() {
 			@Override
@@ -302,6 +304,22 @@ public void fromFormDataWith() throws Exception {
 
 	}
 
+	@Test
+	public void fromMultipartData() throws Exception {
+		MultiValueMap<String, Object> map = new LinkedMultiValueMap<>();
+		map.set("name 3", "value 3");
+
+		BodyInserters.FormInserter<Object> inserter =
+				BodyInserters.fromMultipartData("name 1", "value 1")
+						.withPublisher("name 2", Flux.just("foo", "bar", "baz"), String.class)
+						.with(map);
+
+		MockClientHttpRequest request = new MockClientHttpRequest(HttpMethod.GET, URI.create("http://example.com"));
+		Mono<Void> result = inserter.insert(request, this.context);
+		StepVerifier.create(result).expectComplete().verify();
+
+	}
+
 	@Test
 	public void ofDataBuffers() throws Exception {
 		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();