Use specified ResolvableType in JacksonJsonEncoder

This commit also fixes an issue in the HTTP client that used the
wrapper type instead of the element type. As a consequence, due
to type erasure, we now have to specify the type of the content
in DefaultHttpRequestBuilder#contentStream().

spring-web-reactive/src/main/java/org/springframework/core/codec/support/JacksonJsonEncoder.java

@@ -21,7 +21,10 @@ import java.io.OutputStream;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 
+import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.databind.type.TypeFactory;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -66,10 +69,13 @@ public class JacksonJsonEncoder extends AbstractEncoder<Object> {
 	public Flux<DataBuffer> encode(Publisher<?> inputStream,
 			DataBufferFactory bufferFactory, ResolvableType elementType, MimeType mimeType,
 			Object... hints) {
+		Assert.notNull(inputStream, "'inputStream' must not be null");
+		Assert.notNull(bufferFactory, "'bufferFactory' must not be null");
+		Assert.notNull(elementType, "'elementType' must not be null");
 		if (inputStream instanceof Mono) {
 			// single object
 			return Flux.from(inputStream)
-					.map(value -> serialize(value, bufferFactory));
+					.map(value -> serialize(value, bufferFactory, elementType));
 		}
 		else {
 			// array
@@ -81,7 +87,7 @@ public class JacksonJsonEncoder extends AbstractEncoder<Object> {
 					Mono.just(bufferFactory.wrap(END_ARRAY_BUFFER));
 
 			Flux<DataBuffer> serializedObjects = Flux.from(inputStream)
-					.map(value -> serialize(value, bufferFactory));
+					.map(value -> serialize(value, bufferFactory, elementType));
 
 			Flux<DataBuffer> array = Flux.zip(serializedObjects, arraySeparators)
 					.flatMap(tuple -> Flux.just(tuple.getT1(), tuple.getT2()));
@@ -92,11 +98,15 @@ public class JacksonJsonEncoder extends AbstractEncoder<Object> {
 		}
 	}
 
-	private DataBuffer serialize(Object value, DataBufferFactory dataBufferFactory) {
+	private DataBuffer serialize(Object value, DataBufferFactory dataBufferFactory,
+			ResolvableType type) {
+		TypeFactory typeFactory = this.mapper.getTypeFactory();
+		JavaType javaType = typeFactory.constructType(type.getType());
+		ObjectWriter writer = this.mapper.writerFor(javaType);
 		DataBuffer buffer = dataBufferFactory.allocateBuffer();
 		OutputStream outputStream = buffer.asOutputStream();
 		try {
-			this.mapper.writeValue(outputStream, value);
+			writer.writeValue(outputStream, value);
 		}
 		catch (IOException e) {
 			throw new CodecException("Error while writing the data", e);

spring-web-reactive/src/main/java/org/springframework/web/client/reactive/DefaultHttpRequestBuilder.java

@@ -60,6 +60,8 @@ public class DefaultHttpRequestBuilder implements HttpRequestBuilder {
 
 	protected Publisher contentPublisher;
 
+	protected ResolvableType contentType;
+
 	protected final List<HttpCookie> cookies = new ArrayList<HttpCookie>();
 
 	protected DefaultHttpRequestBuilder() {
@@ -111,11 +113,13 @@ public class DefaultHttpRequestBuilder implements HttpRequestBuilder {
 
 	public DefaultHttpRequestBuilder content(Object content) {
 		this.contentPublisher = Mono.just(content);
+		this.contentType = ResolvableType.forInstance(content);
 		return this;
 	}
 
-	public DefaultHttpRequestBuilder contentStream(Publisher<?> content) {
+	public DefaultHttpRequestBuilder contentStream(Publisher<?> content, ResolvableType type) {
 		this.contentPublisher = Flux.from(content);
+		this.contentType = type;
 		return this;
 	}
 
@@ -139,22 +143,21 @@ public class DefaultHttpRequestBuilder implements HttpRequestBuilder {
 		request.getHeaders().putAll(this.httpHeaders);
 
 		if (this.contentPublisher != null) {
-			ResolvableType requestBodyType = ResolvableType.forInstance(this.contentPublisher);
 			MediaType mediaType = request.getHeaders().getContentType();
 
 			Optional<Encoder<?>> messageEncoder = messageEncoders
 					.stream()
-					.filter(e -> e.canEncode(requestBodyType, mediaType))
+					.filter(e -> e.canEncode(this.contentType, mediaType))
 					.findFirst();
 
 			if (messageEncoder.isPresent()) {
 				request.writeWith(messageEncoder.get()
 						.encode(this.contentPublisher, request.bufferFactory(),
-								requestBodyType, mediaType));
+								this.contentType, mediaType));
 			}
 			else {
 				throw new WebClientException("Can't write request body " +
-						"of type '" + requestBodyType.toString() +
+						"of type '" + this.contentType.toString() +
 						"' for content-type '" + mediaType.toString() + "'");
 			}
 		}

spring-web-reactive/src/test/java/org/springframework/core/codec/support/JacksonJsonEncoderTests.java

@@ -16,11 +16,14 @@
 
 package org.springframework.core.codec.support;
 
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.annotation.JsonTypeName;
 import org.junit.Before;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.test.TestSubscriber;
 
+import org.springframework.core.ResolvableType;
 import org.springframework.core.io.buffer.AbstractDataBufferAllocatingTestCase;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.MediaType;
@@ -50,8 +53,9 @@ public class JacksonJsonEncoderTests extends AbstractDataBufferAllocatingTestCas
 	public void write() {
 		Flux<Pojo> source = Flux.just(new Pojo("foofoo", "barbar"), new Pojo("foofoofoo", "barbarbar"));
 
+		ResolvableType type = ResolvableType.forClass(Pojo.class);
 		Flux<DataBuffer> output =
-				this.encoder.encode(source, this.dataBufferFactory, null, null);
+				this.encoder.encode(source, this.dataBufferFactory, type, null);
 
 		TestSubscriber
 				.subscribe(output)
@@ -64,4 +68,35 @@ public class JacksonJsonEncoderTests extends AbstractDataBufferAllocatingTestCas
 						stringConsumer("]"));
 	}
 
+	@Test
+	public void writeWithType() {
+		Flux<ParentClass> source = Flux.just(new Foo(), new Bar());
+
+		ResolvableType type = ResolvableType.forClass(ParentClass.class);
+		Flux<DataBuffer> output =
+				this.encoder.encode(source, this.dataBufferFactory, type, null);
+
+		TestSubscriber
+				.subscribe(output)
+				.assertComplete()
+				.assertNoError()
+				.assertValuesWith(stringConsumer("["),
+						stringConsumer("{\"type\":\"foo\"}"),
+						stringConsumer(","),
+						stringConsumer("{\"type\":\"bar\"}"),
+						stringConsumer("]"));
+	}
+
+	@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.PROPERTY, property = "type")
+	private static class ParentClass {
+	}
+
+	@JsonTypeName("foo")
+	private static class Foo extends ParentClass {
+	}
+
+	@JsonTypeName("bar")
+	private static class Bar extends ParentClass {
+	}
+
 }