INT-3548 Fix @Payloads Annotation

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

Refactoring in 4.1 broke the `@Payloads` annotation in the
`MessagingMethodInvokerHelper`.

The test case missed this because it explicitly declares
`@Payloads List<Integer>` which is actually handled by the
case below:

````java
else if (Collection.class.isAssignableFrom(parameterType) || parameterType.isArray()) {
	if (canProcessMessageList) {
		sb.append("messages.![payload]");
		...
````

`List<?>` triggered

````java
else if ((parameterTypeDescriptor.isAssignableTo(messageListTypeDescriptor) || parameterTypeDescriptor
				.isAssignableTo(messageArrayTypeDescriptor))) {
	sb.append("messages");
	...
````

- Add back the logic to detect `@Payloads`.
- Remove `@Payloads` from the existing test.
- Add a new test that uses `List<?>`

spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayMethodInboundMessageMapper.java

@@ -139,6 +139,7 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
 		this.payloadExpression = PARSER.parseExpression(expressionString);
 	}
 
+	@Override
 	public void setBeanFactory(final BeanFactory beanFactory) {
 		if (beanFactory != null) {
 			this.beanFactory = beanFactory;
@@ -146,7 +147,8 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
 		}
 	}
 
- 	public Message<?> toMessage(Object[] arguments) {
+	@Override
+	public Message<?> toMessage(Object[] arguments) {
 		Assert.notNull(arguments, "cannot map null arguments to Message");
 		if (arguments.length != this.parameterList.size()) {
 			String prefix = (arguments.length < this.parameterList.size()) ? "Not enough" : "Too many";
@@ -275,7 +277,7 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
 				Object argumentValue = arguments[i];
 				MethodParameter methodParameter = GatewayMethodInboundMessageMapper.this.parameterList.get(i);
 				Annotation annotation =
-						MessagingAnnotationUtils.findMessagePartAnnotation(methodParameter.getParameterAnnotations());
+						MessagingAnnotationUtils.findMessagePartAnnotation(methodParameter.getParameterAnnotations(), false);
 				if (annotation != null) {
 					if (annotation.annotationType().equals(org.springframework.integration.annotation.Payload.class)
 							|| annotation.annotationType().equals(Payload.class)) {

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

@@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicReference;
 import org.springframework.aop.support.AopUtils;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.integration.annotation.Payloads;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.Headers;
@@ -97,14 +98,15 @@ public final class MessagingAnnotationUtils {
 
 	/**
 	 * Find the one of {@link Payload}, {@link Header} or {@link Headers} annotation from
-	 * the provided {@code annotations} array.
+	 * the provided {@code annotations} array. Optionally also detects {@link Payloads}.
 	 * @param annotations the annotations to scan.
+	 * @param payloads true if @Payloads should be detected.
 	 * @return the matched annotation or {@code null}.
 	 * @throws MessagingException if more than one of {@link Payload}, {@link Header}
 	 * or {@link Headers} annotations are presented.
 	 */
 	@SuppressWarnings("deprecation")
-	public static Annotation findMessagePartAnnotation(Annotation[] annotations) {
+	public static Annotation findMessagePartAnnotation(Annotation[] annotations, boolean payloads) {
 		if (annotations == null || annotations.length == 0) {
 			return null;
 		}
@@ -116,7 +118,8 @@ public final class MessagingAnnotationUtils {
 					|| type.equals(org.springframework.integration.annotation.Header.class)
 					|| type.equals(Header.class)
 					|| type.equals(org.springframework.integration.annotation.Headers.class)
-					|| type.equals(Headers.class)) {
+					|| type.equals(Headers.class)
+					|| (payloads && type.equals(Payloads.class))) {
 				if (match != null) {
 					throw new MessagingException("At most one parameter annotation can be provided "
 							+ "for message mapping, but found two: [" + match.annotationType().getName() + "] and ["

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

@@ -605,7 +605,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 				TypeDescriptor parameterTypeDescriptor = new TypeDescriptor(methodParameter);
 				Class<?> parameterType = parameterTypeDescriptor.getObjectType();
 				Annotation mappingAnnotation =
-						MessagingAnnotationUtils.findMessagePartAnnotation(parameterAnnotations[i]);
+						MessagingAnnotationUtils.findMessagePartAnnotation(parameterAnnotations[i], true);
 				if (mappingAnnotation != null) {
 					Class<? extends Annotation> annotationType = mappingAnnotation.annotationType();
 					if (annotationType.equals(org.springframework.integration.annotation.Payload.class)

spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java

@@ -56,7 +56,7 @@ import org.springframework.messaging.support.GenericMessage;
 @RunWith(MockitoJUnitRunner.class)
 public class MethodInvokingMessageGroupProcessorTests {
 
-	private List<Message<?>> messagesUpForProcessing = new ArrayList<Message<?>>(3);
+	private final List<Message<?>> messagesUpForProcessing = new ArrayList<Message<?>>(3);
 
 	@Mock
 	private MessageGroup messageGroupMock;
@@ -137,11 +137,37 @@ public class MethodInvokingMessageGroupProcessorTests {
 	}
 
 	@Test
-	public void shouldFindAnnotatedPayloads() throws Exception {
+	public void shouldFindListPayloads() throws Exception {
 
 		@SuppressWarnings("unused")
 		class SimpleAggregator {
-			public String and(@Payloads List<Integer> flags, @Header("foo") List<Integer> header) {
+			public String and(List<Integer> flags, @Header("foo") List<Integer> header) {
+				List<Integer> result = new ArrayList<Integer>();
+				for (int flag : flags) {
+					result.add(flag);
+				}
+				for (int flag : header) {
+					result.add(flag);
+				}
+				return result.toString();
+			}
+		}
+
+		MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
+		messagesUpForProcessing.add(MessageBuilder.withPayload(3).setHeader("foo", Arrays.asList(101, 102)).build());
+		when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing);
+		Object result = processor.processMessageGroup(messageGroupMock);
+		assertThat((String) ((Message<?>) result).getPayload(), is("[1, 2, 4, 3, 101, 102]"));
+	}
+
+	@Test
+	public void shouldFindAnnotatedPayloadsWithNoType() throws Exception {
+
+		@SuppressWarnings("unused")
+		class SimpleAggregator {
+			public String and(@Payloads List<?> rawFlags, @Header("foo") List<Integer> header) {
+				@SuppressWarnings("unchecked")
+				List<Integer> flags = (List<Integer>) rawFlags;
 				List<Integer> result = new ArrayList<Integer>();
 				for (int flag : flags) {
 					result.add(flag);
@@ -242,6 +268,7 @@ public class MethodInvokingMessageGroupProcessorTests {
 		MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator());
 		GenericConversionService conversionService = new DefaultConversionService();
 		conversionService.addConverter(new Converter<ArrayList<?>, Iterator<?>>() {
+			@Override
 			public Iterator<?> convert(ArrayList<?> source) {
 				return source.iterator();
 			}
@@ -523,6 +550,7 @@ public class MethodInvokingMessageGroupProcessorTests {
 			this.greeting = greeting;
 		}
 
+		@Override
 		@Aggregator
 		public String sayHello(List<String> names) {
 			return greeting + " " + names.get(0);