Add ResolvableType to HttpEntity for multipart Publishers

This commit adds a ResolvableType field to HttpEntity, in order to
support Publishers as multipart data. Without the type, the
MultipartHttpMessageWriter does not know which delegate writer to use to
write the part.

Issue: SPR-16307

spring-web/src/main/java/org/springframework/http/HttpEntity.java

@@ -16,7 +16,12 @@
 
 package org.springframework.http;
 
+import org.reactivestreams.Publisher;
+
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.ResolvableType;
 import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
 import org.springframework.util.MultiValueMap;
 import org.springframework.util.ObjectUtils;
 
@@ -67,6 +72,9 @@ public class HttpEntity<T> {
 	@Nullable
 	private final T body;
 
+	@Nullable
+	private final ResolvableType bodyType;
+
 
 	/**
 	 * Create a new, empty {@code HttpEntity}.
@@ -97,7 +105,18 @@ public class HttpEntity<T> {
 	 * @param headers the entity headers
 	 */
 	public HttpEntity(@Nullable T body, @Nullable MultiValueMap<String, String> headers) {
+		this(body, null, headers);
+	}
+
+	private HttpEntity(@Nullable T body, @Nullable ResolvableType bodyType,
+			@Nullable MultiValueMap<String, String> headers) {
 		this.body = body;
+
+		if (bodyType == null && body != null) {
+			bodyType = ResolvableType.forClass(body.getClass());
+		}
+		this.bodyType = bodyType ;
+
 		HttpHeaders tempHeaders = new HttpHeaders();
 		if (headers != null) {
 			tempHeaders.putAll(headers);
@@ -128,6 +147,13 @@ public class HttpEntity<T> {
 		return (this.body != null);
 	}
 
+	/**
+	 * Returns the type of the body.
+	 */
+	@Nullable
+	public ResolvableType getBodyType() {
+		return this.bodyType;
+	}
 
 	@Override
 	public boolean equals(@Nullable Object other) {
@@ -159,4 +185,44 @@ public class HttpEntity<T> {
 		return builder.toString();
 	}
 
+
+	// Static builder methods
+
+	/**
+	 * Create a new {@code HttpEntity} with the given {@link Publisher} as body, class contained in
+	 * {@code publisher}, and headers.
+	 * @param publisher the publisher to use as body
+	 * @param elementClass the class of elements contained in the publisher
+	 * @param headers the entity headers
+	 * @param <S> the type of the elements contained in the publisher
+	 * @param <P> the type of the {@code Publisher}
+	 * @return the created entity
+	 */
+	public static <S, P extends Publisher<S>> HttpEntity<P> fromPublisher(P publisher,
+			Class<S> elementClass, @Nullable MultiValueMap<String, String> headers) {
+
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(elementClass, "'elementClass' must not be null");
+		return new HttpEntity<>(publisher, ResolvableType.forClass(elementClass), headers);
+	}
+
+	/**
+	 * Create a new {@code HttpEntity} with the given {@link Publisher} as body, type contained in
+	 * {@code publisher}, and headers.
+	 * @param publisher the publisher to use as body
+	 * @param typeReference the type of elements contained in the publisher
+	 * @param headers the entity headers
+	 * @param <S> the type of the elements contained in the publisher
+	 * @param <P> the type of the {@code Publisher}
+	 * @return the created entity
+	 */
+	public static <S, P extends Publisher<S>> HttpEntity<P> fromPublisher(P publisher,
+			ParameterizedTypeReference<S> typeReference,
+			@Nullable MultiValueMap<String, String> headers) {
+
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(typeReference, "'typeReference' must not be null");
+		return new HttpEntity<>(publisher, ResolvableType.forType(typeReference), headers);
+	}
+
 }

spring-web/src/main/java/org/springframework/http/client/MultipartBodyBuilder.java

@@ -20,6 +20,10 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 
+import org.reactivestreams.Publisher;
+
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.ResolvableType;
 import org.springframework.http.HttpEntity;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.MediaType;
@@ -96,6 +100,11 @@ public final class MultipartBodyBuilder {
 		Assert.hasLength(name, "'name' must not be empty");
 		Assert.notNull(part, "'part' must not be null");
 
+		if (part instanceof Publisher) {
+			throw new IllegalArgumentException("Use publisher(String, Publisher, Class) or " +
+				"publisher(String, Publisher, ParameterizedTypeReference) for adding Publisher parts");
+		}
+
 		Object partBody;
 		HttpHeaders partHeaders = new HttpHeaders();
 
@@ -116,6 +125,54 @@ public final class MultipartBodyBuilder {
 		return builder;
 	}
 
+	/**
+	 * Adds a {@link Publisher} part to this builder, allowing for further header customization with
+	 * the returned {@link PartBuilder}.
+	 * @param name the name of the part to add (may not be empty)
+	 * @param publisher the contents of the part to add
+	 * @param elementClass the class of elements contained in the publisher
+	 * @return a builder that allows for further header customization
+	 */
+	public <T, P extends Publisher<T>> PartBuilder asyncPart(String name, P publisher,
+			Class<T> elementClass) {
+
+		Assert.notNull(elementClass, "'elementClass' must not be null");
+		ResolvableType elementType = ResolvableType.forClass(elementClass);
+		Assert.hasLength(name, "'name' must not be empty");
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(elementType, "'elementType' must not be null");
+
+		HttpHeaders partHeaders = new HttpHeaders();
+		PublisherClassPartBuilder<T, P> builder =
+				new PublisherClassPartBuilder<>(publisher, elementClass, partHeaders);
+		this.parts.add(name, builder);
+		return builder;
+
+	}
+
+	/**
+	 * Adds a {@link Publisher} part to this builder, allowing for further header customization with
+	 * the returned {@link PartBuilder}.
+	 * @param name the name of the part to add (may not be empty)
+	 * @param publisher the contents of the part to add
+	 * @param elementType the type of elements contained in the publisher
+	 * @return a builder that allows for further header customization
+	 */
+	public <T, P extends Publisher<T>> PartBuilder asyncPart(String name, P publisher,
+			ParameterizedTypeReference<T> elementType) {
+
+		Assert.notNull(elementType, "'elementType' must not be null");
+		ResolvableType elementType1 = ResolvableType.forType(elementType);
+		Assert.hasLength(name, "'name' must not be empty");
+		Assert.notNull(publisher, "'publisher' must not be null");
+		Assert.notNull(elementType1, "'elementType' must not be null");
+
+		HttpHeaders partHeaders = new HttpHeaders();
+		PublisherTypReferencePartBuilder<T, P> builder =
+				new PublisherTypReferencePartBuilder<>(publisher, elementType, partHeaders);
+		this.parts.add(name, builder);
+		return builder;
+	}
 
 	/**
 	 * Builder interface that allows for customization of part headers.
@@ -136,10 +193,9 @@ public final class MultipartBodyBuilder {
 	private static class DefaultPartBuilder implements PartBuilder {
 
 		@Nullable
-		private final Object body;
-
-		private final HttpHeaders headers;
+		protected final Object body;
 
+		protected final HttpHeaders headers;
 
 		public DefaultPartBuilder(@Nullable Object body, HttpHeaders headers) {
 			this.body = body;
@@ -157,4 +213,44 @@ public final class MultipartBodyBuilder {
 		}
 	}
 
+	private static class PublisherClassPartBuilder<S, P extends Publisher<S>>
+			extends DefaultPartBuilder {
+
+		private final Class<S> bodyType;
+
+		public PublisherClassPartBuilder(P body, Class<S> bodyType, HttpHeaders headers) {
+			super(body, headers);
+			this.bodyType = bodyType;
+		}
+
+		@Override
+		@SuppressWarnings("unchecked")
+		public HttpEntity<?> build() {
+			P body = (P) this.body;
+			Assert.state(body != null, "'body' must not be null");
+			return HttpEntity.fromPublisher(body, this.bodyType, this.headers);
+		}
+	}
+
+	private static class PublisherTypReferencePartBuilder<S, P extends Publisher<S>>
+			extends DefaultPartBuilder {
+
+		private final ParameterizedTypeReference<S> bodyType;
+
+		public PublisherTypReferencePartBuilder(P body, ParameterizedTypeReference<S> bodyType,
+				HttpHeaders headers) {
+
+			super(body, headers);
+			this.bodyType = bodyType;
+		}
+
+		@Override
+		@SuppressWarnings("unchecked")
+		public HttpEntity<?> build() {
+			P body = (P) this.body;
+			Assert.state(body != null, "'body' must not be null");
+			return HttpEntity.fromPublisher(body, this.bodyType, this.headers);
+		}
+	}
+
 }

spring-web/src/main/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriter.java

@@ -230,31 +230,41 @@ public class MultipartHttpMessageWriter implements HttpMessageWriter<MultiValueM
 		MultipartHttpOutputMessage outputMessage = new MultipartHttpOutputMessage(this.bufferFactory, getCharset());
 
 		T body;
+		ResolvableType bodyType = null;
 		if (value instanceof HttpEntity) {
-			outputMessage.getHeaders().putAll(((HttpEntity<T>) value).getHeaders());
-			body = ((HttpEntity<T>) value).getBody();
+			HttpEntity<T> httpEntity = (HttpEntity<T>) value;
+			outputMessage.getHeaders().putAll(httpEntity.getHeaders());
+			body = httpEntity.getBody();
 			Assert.state(body != null, "MultipartHttpMessageWriter only supports HttpEntity with body");
+			bodyType = httpEntity.getBodyType();
 		}
 		else {
 			body = value;
 		}
 
+		if (bodyType == null) {
+			bodyType = ResolvableType.forClass(body.getClass());
+		}
+
 		String filename = (body instanceof Resource ? ((Resource) body).getFilename() : null);
 		outputMessage.getHeaders().setContentDispositionFormData(name, filename);
 
-		ResolvableType bodyType = ResolvableType.forClass(body.getClass());
 		MediaType contentType = outputMessage.getHeaders().getContentType();
 
+		final ResolvableType finalBodyType = bodyType;
 		Optional<HttpMessageWriter<?>> writer = this.partWriters.stream()
-				.filter(partWriter -> partWriter.canWrite(bodyType, contentType))
+				.filter(partWriter -> partWriter.canWrite(finalBodyType, contentType))
 				.findFirst();
 
 		if (!writer.isPresent()) {
 			return Flux.error(new CodecException("No suitable writer found for part: " + name));
 		}
 
+		Publisher<T> bodyPublisher =
+				body instanceof Publisher ? (Publisher<T>) body : Mono.just(body);
+
 		Mono<Void> partWritten = ((HttpMessageWriter<T>) writer.get())
-				.write(Mono.just(body), bodyType, contentType, outputMessage, Collections.emptyMap());
+				.write(bodyPublisher, bodyType, contentType, outputMessage, Collections.emptyMap());
 
 		// partWritten.subscribe() is required in order to make sure MultipartHttpOutputMessage#getBody()
 		// returns a non-null value (occurs with ResourceHttpMessageWriter that invokes

spring-web/src/test/java/org/springframework/http/client/MultipartBodyBuilderTests.java

@@ -17,7 +17,10 @@
 package org.springframework.http.client;
 
 import org.junit.Test;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 
+import org.springframework.core.ResolvableType;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
 import org.springframework.http.HttpEntity;
@@ -34,23 +37,25 @@ public class MultipartBodyBuilderTests {
 
 	@Test
 	public void builder() throws Exception {
-		MultiValueMap<String, String> form = new LinkedMultiValueMap<>();
-		form.add("form field", "form value");
+		MultiValueMap<String, String> multipartData = new LinkedMultiValueMap<>();
+		multipartData.add("form field", "form value");
 		Resource logo = new ClassPathResource("/org/springframework/http/converter/logo.jpg");
 		HttpHeaders entityHeaders = new HttpHeaders();
 		entityHeaders.add("foo", "bar");
 		HttpEntity<String> entity = new HttpEntity<>("body", entityHeaders);
+		Publisher<String> publisher = Flux.just("foo", "bar", "baz");
 
 		MultipartBodyBuilder builder = new MultipartBodyBuilder();
-		builder.part("key", form).header("foo", "bar");
+		builder.part("key", multipartData).header("foo", "bar");
 		builder.part("logo", logo).header("baz", "qux");
 		builder.part("entity", entity).header("baz", "qux");
+		builder.asyncPart("publisher", publisher, String.class).header("baz", "qux");
 
 		MultiValueMap<String, HttpEntity<?>> result = builder.build();
 
-		assertEquals(3, result.size());
+		assertEquals(4, result.size());
 		assertNotNull(result.getFirst("key"));
-		assertEquals(form, result.getFirst("key").getBody());
+		assertEquals(multipartData, result.getFirst("key").getBody());
 		assertEquals("bar", result.getFirst("key").getHeaders().getFirst("foo"));
 
 		assertNotNull(result.getFirst("logo"));
@@ -61,6 +66,12 @@ public class MultipartBodyBuilderTests {
 		assertEquals("body", result.getFirst("entity").getBody());
 		assertEquals("bar", result.getFirst("entity").getHeaders().getFirst("foo"));
 		assertEquals("qux", result.getFirst("entity").getHeaders().getFirst("baz"));
+
+		assertNotNull(result.getFirst("publisher"));
+		assertEquals(publisher, result.getFirst("publisher").getBody());
+		assertEquals(ResolvableType.forClass(String.class), result.getFirst("publisher").getBodyType());
+		assertEquals("bar", result.getFirst("entity").getHeaders().getFirst("foo"));
+		assertEquals("qux", result.getFirst("entity").getHeaders().getFirst("baz"));
 	}
 
 

spring-web/src/test/java/org/springframework/http/codec/multipart/MultipartHttpMessageWriterTests.java

@@ -22,6 +22,8 @@ import java.util.List;
 import java.util.Map;
 
 import org.junit.Test;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
@@ -36,10 +38,7 @@ import org.springframework.mock.http.server.reactive.test.MockServerHttpRequest;
 import org.springframework.mock.http.server.reactive.test.MockServerHttpResponse;
 import org.springframework.util.MultiValueMap;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 
 /**
  * @author Sebastien Deleuze
@@ -80,6 +79,8 @@ public class MultipartHttpMessageWriterTests {
 			}
 		};
 
+		Publisher<String> publisher = Flux.just("foo", "bar", "baz");
+
 		MultipartBodyBuilder bodyBuilder = new MultipartBodyBuilder();
 		bodyBuilder.part("name 1", "value 1");
 		bodyBuilder.part("name 2", "value 2+1");
@@ -87,14 +88,15 @@ public class MultipartHttpMessageWriterTests {
 		bodyBuilder.part("logo", logo);
 		bodyBuilder.part("utf8", utf8);
 		bodyBuilder.part("json", new Foo("bar"), MediaType.APPLICATION_JSON_UTF8);
-		Mono<MultiValueMap<String, HttpEntity<?>>> publisher = Mono.just(bodyBuilder.build());
+		bodyBuilder.asyncPart("publisher", publisher, String.class);
+		Mono<MultiValueMap<String, HttpEntity<?>>> result = Mono.just(bodyBuilder.build());
 
 		MockServerHttpResponse response = new MockServerHttpResponse();
 		Map<String, Object> hints = Collections.emptyMap();
-		this.writer.write(publisher, null, MediaType.MULTIPART_FORM_DATA, response, hints).block(Duration.ofSeconds(5));
+		this.writer.write(result, null, MediaType.MULTIPART_FORM_DATA, response, hints).block(Duration.ofSeconds(5));
 
 		MultiValueMap<String, Part> requestParts = parse(response, hints);
-		assertEquals(5, requestParts.size());
+		assertEquals(6, requestParts.size());
 
 		Part part = requestParts.getFirst("name 1");
 		assertTrue(part instanceof FormFieldPart);
@@ -136,6 +138,17 @@ public class MultipartHttpMessageWriterTests {
 
 		assertEquals("{\"bar\":\"bar\"}", value);
 
+		part = requestParts.getFirst("publisher");
+		assertEquals("publisher", part.name());
+
+		value = StringDecoder.textPlainOnly(false).decodeToMono(part.content(),
+				ResolvableType.forClass(String.class), MediaType.TEXT_PLAIN,
+				Collections.emptyMap()).block(Duration.ZERO);
+
+		assertEquals("foobarbaz", value);
+
+
+
 	}
 
 	private MultiValueMap<String, Part> parse(MockServerHttpResponse response, Map<String, Object> hints) {