INT-4410 CollArgResolver only for group processor

JIRA: https://jira.spring.io/browse/INT-4410

The `CollectionArgumentResolver` has been introduced especially for the
cases to work with `MessageGroupProcessor` (an aggregator)
when the payload is a `Collection<Message<?>>`.

This use-case doesn't apply for the general collection parameter use-case.

* Register `CollectionArgumentResolver` only when `listCapable` option.
For all other collection-based use-cases fallback to the standard
`PayloadArgumentResolver` with an appropriate configured `MessageConverter`

Note: this is for backward compatibility.
In the `5.1` we may reconsider to use `MessageConverter` in the
`CollectionArgumentResolver` as well, or just remove it altogether with
an appropriate logic in the `PayloadArgumentResolver`, since this
`CollectionArgumentResolver` solution isn't robust

spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationRegistrar.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2017 the original author or authors.
+ * Copyright 2014-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -463,10 +463,13 @@ public class IntegrationRegistrar implements ImportBeanDefinitionRegistrar, Bean
 		resolvers.add(new RootBeanDefinition(PayloadExpressionArgumentResolver.class));
 		resolvers.add(new RootBeanDefinition(PayloadsArgumentResolver.class));
 		resolvers.add(new RootBeanDefinition(MapArgumentResolver.class));
-		resolvers.add(
-				BeanDefinitionBuilder.genericBeanDefinition(CollectionArgumentResolver.class)
-						.addConstructorArgValue(listCapable)
-						.getBeanDefinition());
+
+		if (listCapable) {
+			resolvers.add(
+					BeanDefinitionBuilder.genericBeanDefinition(CollectionArgumentResolver.class)
+							.addConstructorArgValue(true)
+							.getBeanDefinition());
+		}
 
 		return BeanDefinitionBuilder.genericBeanDefinition(HandlerMethodArgumentResolversHolder.class)
 				.addConstructorArgValue(resolvers)

spring-integration-core/src/main/java/org/springframework/integration/util/MessagingMethodInvokerHelper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -577,17 +577,19 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		PayloadsArgumentResolver payloadsArgumentResolver = new PayloadsArgumentResolver();
 		payloadsArgumentResolver.setBeanFactory(getBeanFactory());
 
-		CollectionArgumentResolver collectionArgumentResolver =
-				new CollectionArgumentResolver(this.canProcessMessageList);
-		collectionArgumentResolver.setBeanFactory(getBeanFactory());
-
 		MapArgumentResolver mapArgumentResolver = new MapArgumentResolver();
 		mapArgumentResolver.setBeanFactory(getBeanFactory());
 
 		List<HandlerMethodArgumentResolver> customArgumentResolvers = new LinkedList<>();
 		customArgumentResolvers.add(payloadExpressionArgumentResolver);
 		customArgumentResolvers.add(payloadsArgumentResolver);
-		customArgumentResolvers.add(collectionArgumentResolver);
+
+		if (this.canProcessMessageList) {
+			CollectionArgumentResolver collectionArgumentResolver = new CollectionArgumentResolver(true);
+			collectionArgumentResolver.setBeanFactory(getBeanFactory());
+			customArgumentResolvers.add(collectionArgumentResolver);
+		}
+
 		customArgumentResolvers.add(mapArgumentResolver);
 
 		this.messageHandlerMethodFactory.setCustomArgumentResolvers(customArgumentResolvers);

spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,9 +31,11 @@ import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.Mockito.mock;
 
 import java.lang.reflect.Method;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.Date;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
@@ -41,6 +43,7 @@ import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
 
 import org.aopalliance.intercept.MethodInterceptor;
 import org.apache.commons.logging.Log;
@@ -56,6 +59,9 @@ import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Configuration;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.SpelCompilerMode;
 import org.springframework.expression.spel.SpelEvaluationException;
@@ -64,6 +70,7 @@ import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.annotation.UseSpelInvoker;
+import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.gateway.GatewayProxyFactoryBean;
 import org.springframework.integration.gateway.RequestReplyExchanger;
 import org.springframework.integration.support.MessageBuilder;
@@ -79,6 +86,9 @@ import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.StopWatch;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 
 /**
  * @author Mark Fisher
@@ -896,7 +906,7 @@ public class MethodInvokingMessageProcessorTests {
 		catch (IllegalArgumentException e) {
 			assertThat(e.getMessage(), equalTo(
 					"UseSpelInvoker.compilerMode: Object of class [java.lang.Object] "
-					+ "must be an instance of class java.lang.String"));
+							+ "must be an instance of class java.lang.String"));
 		}
 
 		// Check other CTORs
@@ -916,7 +926,7 @@ public class MethodInvokingMessageProcessorTests {
 		SingleMethodJsonWithSpELBean bean = new SingleMethodJsonWithSpELBean();
 		MessagingMethodInvokerHelper<?> helper = new MessagingMethodInvokerHelper<>(bean,
 				SingleMethodJsonWithSpELBean.class.getDeclaredMethod("foo",
-				SingleMethodJsonWithSpELBean.Foo.class),
+						SingleMethodJsonWithSpELBean.Foo.class),
 				false);
 		Message<?> message = new GenericMessage<>("{\"bar\":\"bar\"}",
 				Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "application/json"));
@@ -979,6 +989,87 @@ public class MethodInvokingMessageProcessorTests {
 	}
 
 
+	@Test
+	public void testCollectionArgument() throws JsonProcessingException {
+
+		class A {
+
+			@SuppressWarnings("unused")
+			public String myMethod(List<Employee<Person>> msg) {
+				return msg.stream()
+						.map(Employee::getEntity)
+						.map(Person::getName)
+						.collect(Collectors.joining(","));
+			}
+
+		}
+
+		ApplicationContext applicationContext = new AnnotationConfigApplicationContext(TestConfiguration.class);
+
+		MethodInvokingMessageProcessor processor = new MethodInvokingMessageProcessor(new A(), "myMethod");
+		processor.setBeanFactory(applicationContext);
+
+		List<Employee<Person>> testData =
+				Arrays.asList(
+						new Employee<>(new Person("Foo")),
+						new Employee<>(new Person("Bar")));
+
+		ObjectMapper objectMapper = new ObjectMapper();
+		byte[] value = objectMapper.writeValueAsBytes(testData);
+
+		String result = (String) processor.processMessage(new GenericMessage<>(value));
+
+		assertEquals("Foo,Bar", result);
+	}
+
+	public static class Employee<T> {
+
+		private T entity;
+
+		public Employee() {
+		}
+
+		public Employee(T entity) {
+			this.entity = entity;
+		}
+
+		public void setEntity(T entity) {
+			this.entity = entity;
+		}
+
+		public T getEntity() {
+			return this.entity;
+		}
+
+	}
+
+	public static class Person {
+
+		private String name;
+
+		public Person() {
+		}
+
+		public Person(String name) {
+			this.name = name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public String getName() {
+			return this.name;
+		}
+
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class TestConfiguration {
+
+	}
+
 	public interface Foo {
 
 		String handle(String payload);