GH-1098 Fixed argument resolution for typed collections

Resolves #1098

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java

@@ -228,5 +228,4 @@ public class BindingServiceConfiguration {
 			}
 		};
 	}
-
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.stream.converter;
 
 import java.io.IOException;
+import java.lang.reflect.ParameterizedType;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -46,7 +47,6 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 	
 	private final Map<ParameterizedTypeReference<?>, JavaType> typeCache = new ConcurrentHashMap<>();
 	
-
 	@Override
 	protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) {
 		if (payload instanceof byte[]) {
@@ -73,9 +73,13 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 				 */
 				conversionHint = null; 
 			}
+			else if (((MethodParameter)conversionHint).getGenericParameterType() instanceof ParameterizedType) {
+				ParameterizedTypeReference<Object> forType = ParameterizedTypeReference.forType(((MethodParameter)conversionHint).getGenericParameterType());
+				result = convertParameterizedType(message, targetClass, forType);
+			}
 		}
 		else if (conversionHint instanceof ParameterizedTypeReference) {
-			result = convertParameterizedType(message, targetClass, conversionHint);
+			result = convertParameterizedType(message, targetClass, (ParameterizedTypeReference<?>)conversionHint);
 		}
 		
 		if (result == null) {
@@ -90,15 +94,14 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 		return result;
 	}
 	
-	private Object convertParameterizedType(Message<?> message, Class<?> targetClass, Object conversionHint) {
+	private Object convertParameterizedType(Message<?> message, Class<?> targetClass, ParameterizedTypeReference<?> conversionHint) {
 		ObjectMapper objectMapper = this.getObjectMapper();
 		Object payload = message.getPayload();
 		try {
 			JavaType type = this.typeCache.get(conversionHint);
 			if (type == null) {
-				type = objectMapper.getTypeFactory().constructType(
-						((ParameterizedTypeReference<?>) conversionHint).getType());
-				this.typeCache.put((ParameterizedTypeReference<?>) conversionHint, type);
+				type = objectMapper.getTypeFactory().constructType((conversionHint).getType());
+				this.typeCache.put(conversionHint, type);
 			}
 			if (payload instanceof byte[]) {
 				return objectMapper.readValue((byte[]) payload, type);

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.binder.tck;
 
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
+import java.util.List;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -51,6 +52,7 @@ import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
@@ -381,6 +383,31 @@ public class ContentTypeTckTests {
 		assertTrue(binder.getLastError().getPayload() instanceof MessageConversionException);
 	}
 	
+	
+	@Test
+	public void toCollectionWithParameterizedType() throws Exception {	
+		ApplicationContext context = new SpringApplicationBuilder(CollectionWithParameterizedTypes.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false");
+		SourceDestination source = context.getBean(SourceDestination.class);
+		TargetDestination target = context.getBean(TargetDestination.class);
+		String jsonPayload = "[{\"person\":{\"name\":\"jon\"},\"id\":123},{\"person\":{\"name\":\"jane\"},\"id\":456}]";
+		source.send(new GenericMessage<byte[]>(jsonPayload.getBytes()));
+		Message<byte[]> outputMessage = target.receive();
+		assertThat(outputMessage.getPayload()).isEqualTo(jsonPayload.getBytes());
+	}
+	
+	@EnableBinding(Processor.class)
+	@Import(SpringIntegrationBinderConfiguration.class)
+	public static class CollectionWithParameterizedTypes {
+		@StreamListener(Processor.INPUT)
+		@SendTo(Processor.OUTPUT)
+		public List<Employee<Person>> echo(List<Employee<Person>> value) {
+			assertTrue(value.get(0) instanceof Employee);
+			return value;
+		}
+	}
+	
 	@EnableBinding(Processor.class)
 	@Import(SpringIntegrationBinderConfiguration.class)
 	public static class TextInJsonOutListener {
@@ -391,8 +418,6 @@ public class ContentTypeTckTests {
 		}
 	}
 	
-	
-	
 	@EnableBinding(Processor.class)
 	@Import(SpringIntegrationBinderConfiguration.class)
 	public static class PojoToPojoStreamListener {
@@ -508,6 +533,23 @@ public class ContentTypeTckTests {
 		}
 	}
 	
+	public static class Employee<P> {
+		private P person;
+		private int id;
+		public int getId() {
+			return id;
+		}
+		public void setId(int id) {
+			this.id = id;
+		}
+		public P getPerson() {
+			return person;
+		}
+		public void setPerson(P person) {
+			this.person = person;
+		}
+	}
+	
 	public static class Person {
 		private String name;