passes a conversion hint to the message converter to support nested generics. Fixes #1723

Resolves #1725

spring-cloud-stream-reactive/pom.xml

@@ -35,6 +35,11 @@
 			<artifactId>spring-cloud-stream-test-support-internal</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+		    <groupId>io.projectreactor</groupId>
+		    <artifactId>reactor-test</artifactId>
+		    <scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-autoconfigure-processor</artifactId>

spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapter.java

@@ -20,6 +20,7 @@ import reactor.core.publisher.Flux;
 
 import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
 import org.springframework.core.MethodParameter;
+import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.ResolvableType;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -66,6 +67,9 @@ public class MessageChannelToInputFluxParameterAdapter
 
 		final Object monitor = new Object();
 
+		Object conversionHint = ParameterizedTypeReference
+				.forType(fluxTypeParameter.getType());
+
 		if (Message.class.isAssignableFrom(fluxTypeParameterClass)) {
 
 			final ResolvableType payloadTypeParameter = fluxTypeParameter.getGeneric(0);
@@ -85,7 +89,7 @@ public class MessageChannelToInputFluxParameterAdapter
 						else {
 							emitter.next(MessageBuilder.createMessage(
 									this.messageConverter.fromMessage(message,
-											payloadTypeParameterClass),
+											payloadTypeParameterClass, conversionHint),
 									message.getHeaders()));
 						}
 					}
@@ -104,7 +108,7 @@ public class MessageChannelToInputFluxParameterAdapter
 						}
 						else {
 							emitter.next(this.messageConverter.fromMessage(message,
-									fluxTypeParameterClass));
+									fluxTypeParameterClass, conversionHint));
 						}
 					}
 				};

spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/converter/TestApplicationJsonMessageMarshallingConverter.java

@@ -0,0 +1,26 @@
+package org.springframework.cloud.stream.converter;
+
+import org.springframework.cloud.stream.reactive.MessageChannelToInputFluxParameterAdapterTests;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * This class exists because the {@link ApplicationJsonMessageMarshallingConverter} is package-private scoped, 
+ * so it can't be used directly in, e.g. the {@link MessageChannelToInputFluxParameterAdapterTests} class.
+ * In that case, we are just extending it and making it public (but only for tests).
+ */
+public class TestApplicationJsonMessageMarshallingConverter
+		extends ApplicationJsonMessageMarshallingConverter {
+
+	public TestApplicationJsonMessageMarshallingConverter() {
+		this(new ObjectMapper().configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false)
+				.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false));
+	}
+
+	public TestApplicationJsonMessageMarshallingConverter(ObjectMapper objectMapper) {
+		super(objectMapper);
+	}
+
+}

spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/MessageChannelToInputFluxParameterAdapterTests.java

@@ -16,17 +16,20 @@
 
 package org.springframework.cloud.stream.reactive;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
-import reactor.core.publisher.Flux;
-
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.converter.TestApplicationJsonMessageMarshallingConverter;
 import org.springframework.core.MethodParameter;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.messaging.Message;
@@ -35,7 +38,8 @@ import org.springframework.messaging.converter.MappingJackson2MessageConverter;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.ReflectionUtils;
 
-import static org.assertj.core.api.Assertions.assertThat;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
 
 /**
  * @author Marius Bogoevici
@@ -77,8 +81,152 @@ public class MessageChannelToInputFluxParameterAdapterTests {
 
 	}
 
+	@Test
+	public void testAdapterConvertsUsingConversionHint() {
+		CompositeMessageConverter messageConverter = new CompositeMessageConverter(
+				Collections
+						.singleton(new TestApplicationJsonMessageMarshallingConverter()));
+
+		MessageChannelToInputFluxParameterAdapter adapter = new MessageChannelToInputFluxParameterAdapter(
+				messageConverter);
+
+		Method processMethod = ReflectionUtils.findMethod(
+				MessageChannelToInputFluxParameterAdapterTests.class, "processNestedGenericFlux",
+				Flux.class);
+
+		DirectChannel adaptedChannel = new DirectChannel();
+
+		@SuppressWarnings("unchecked")
+		Flux<FirstLevelWrapper<SecondLevelWrapper<String>>> adapterFlux = (Flux<FirstLevelWrapper<SecondLevelWrapper<String>>>) adapter
+				.adapt(adaptedChannel, new MethodParameter(processMethod, 0));
+
+		SecondLevelWrapper<String> expected2 = new SecondLevelWrapper<>();
+		expected2.setName("name");
+		expected2.setData("data");
+
+		FirstLevelWrapper<SecondLevelWrapper<String>> expected1 = new FirstLevelWrapper<>();
+		expected1.setId(1);
+		expected1.setData(expected2);
+
+		StepVerifier.create(adapterFlux).then(() -> {
+			adaptedChannel.send(MessageBuilder.withPayload(
+					"{ \"id\": 1, \"data\": { \"name\": \"name\", \"data\": \"data\" } }")
+					.build());
+		}).expectNext(expected1).thenCancel().verify();
+	}
+
 	public void process(Flux<Message<?>> message) {
 		// do nothing - we just reference this method from the test
 	}
 
+	public void processNestedGenericFlux(Flux<FirstLevelWrapper<SecondLevelWrapper<String>>> items) {
+		// do nothing - we just reference this method from the test
+	}
+
+	static class FirstLevelWrapper<T> {
+
+		private long id;
+
+		private T data;
+
+		long getId() {
+			return id;
+		}
+
+		void setId(long id) {
+			this.id = id;
+		}
+
+		T getData() {
+			return data;
+		}
+
+		public void setData(T data) {
+			this.data = data;
+		}
+
+		@Override
+		public int hashCode() {
+			return Objects.hash(data, id);
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (this == obj) {
+				return true;
+			}
+
+			if (obj == null) {
+				return false;
+			}
+
+			if (!(obj instanceof FirstLevelWrapper)) {
+				return false;
+			}
+
+			@SuppressWarnings("unchecked")
+			FirstLevelWrapper<T> other = (FirstLevelWrapper<T>) obj;
+			return Objects.equals(data, other.data) && id == other.id;
+		}
+
+		@Override
+		public String toString() {
+			return "FirstLevelWrapper [id=" + id + ", data=" + data + "]";
+		}
+
+	}
+
+	static class SecondLevelWrapper<T> {
+
+		private String name;
+
+		private T data;
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public T getData() {
+			return data;
+		}
+
+		public void setData(T data) {
+			this.data = data;
+		}
+
+		@Override
+		public int hashCode() {
+			return Objects.hash(data, name);
+		}
+
+		@Override
+		public boolean equals(Object obj) {
+			if (this == obj) {
+				return true;
+			}
+
+			if (obj == null) {
+				return false;
+			}
+
+			if (!(obj instanceof SecondLevelWrapper)) {
+				return false;
+			}
+
+			@SuppressWarnings("unchecked")
+			SecondLevelWrapper<T> other = (SecondLevelWrapper<T>) obj;
+			return Objects.equals(data, other.data) && Objects.equals(name, other.name);
+		}
+
+		@Override
+		public String toString() {
+			return "SecondLevelWrapper [name=" + name + ", data=" + data + "]";
+		}
+
+	}
+
 }