INT-4214: Migrate to InvocableHandlerMethod

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

* Rework `MessagingMethodInvokerHelper` to delegate to the `InvocableHandlerMethod` with its `HandlerMethodArgumentResolver` infrastructure instead of SpEL
* Introduce several `HandlerMethodArgumentResolver` to address some SI use-cases like `@Payloads`, `@Payload(expression = "")` and `Collection` as argument
* Initialize `DefaultMessageHandlerMethodFactory` in the `MessagingMethodInvokerHelper.start()`.
With that I observed several `Lifecycle` problem when we don't have proper delegate from the top.
* Fix `AbstractCorrelatingMessageHandler` and similar to delegate `Lifecycle` properly
* Fix `ReactiveConsumer` to delegate `Lifecycle` to the `MessageHandler`
* Fix `MutableMessage` do not `generateId()` and set `timestamp` headers if we already have them in the provided headers
* With all that `Lifecycle` many tests must be fixed to call `start()`

Add SpEL fallback variant to the MessagingMethodInvokerHelper

Add `MessagingMethodInvokerHelper.setUseSpelInvoker(boolean)`

Add `MethodInvokingMessageProcessorTests.testPerformanceSpelVersusInvocable()`

Add Compiled SpEL comparison

* Add `MapArgumentResolver` to cover `Properties` case
* add `MessagingMethodInvokerHelper.HandlerMethod.spelOnly` state, when we definitely can perform ony SpEL for provided arguments, e.g. `@Header` with expression
* Catch `IllegalStateException` with the `"argument type mismatch"` message to fallback to SpEL invocation
* Add `Iterator` support for the `CollectionArgumentResolver`
* Add `integrationConversionService` bean registration into the `TestUtils.createTestApplicationContext()`
* Tweak `.travis.yml` to try to download latest JDK, the current `1.8.0_31` is pretty old already and has some bugs
* Adjust some failing tests to use `TestUtils.createTestApplicationContext()` to rely on newly added `integrationConversionService` bean

Fix failed tests: `ctx.refresh()`

Polishing current year in Copyright

* JavaDocs for `CollectionArgumentResolver` and `MapArgumentResolver`
* Propagate `ConversionService` from the `MessagingMethodInvokerHelper` to the `MapArgumentResolver.java`

Fix `MessagingMethodInvokerHelper` to propagate `BeanFactory` into `MapArgumentResolver` as well
The `MapArgumentResolver` now `extends AbstractExpressionEvaluator`, too
Fix JavaDoc typo

spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -37,6 +37,7 @@ import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.context.Lifecycle;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -85,7 +86,7 @@ import org.springframework.util.CollectionUtils;
  * @since 2.0
  */
 public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageProducingHandler
-		implements DiscardingMessageHandler, DisposableBean, ApplicationEventPublisherAware {
+		implements DiscardingMessageHandler, DisposableBean, ApplicationEventPublisherAware, Lifecycle {
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
@@ -131,6 +132,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 
 	private volatile boolean expireGroupsUponTimeout = true;
 
+	private volatile boolean running;
+
 	public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store,
 			CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) {
 		Assert.notNull(processor, "'processor' must not be null");
@@ -750,6 +753,37 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 		}
 	}
 
+	@Override
+	public void start() {
+		if (!this.running) {
+			this.running = true;
+			if (this.outputProcessor instanceof Lifecycle) {
+				((Lifecycle) this.outputProcessor).start();
+			}
+			if (this.releaseStrategy instanceof Lifecycle) {
+				((Lifecycle) this.releaseStrategy).start();
+			}
+		}
+	}
+
+	@Override
+	public void stop() {
+		if (this.running) {
+			this.running = false;
+			if (this.outputProcessor instanceof Lifecycle) {
+				((Lifecycle) this.outputProcessor).stop();
+			}
+			if (this.releaseStrategy instanceof Lifecycle) {
+				((Lifecycle) this.releaseStrategy).stop();
+			}
+		}
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.running;
+	}
+
 	protected static class SequenceAwareMessageGroup extends SimpleMessageGroup {
 
 		private final SimpleMessageGroup sourceGroup;
@@ -787,7 +821,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
 				Integer messageSequenceSize = message.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE,
 						Integer.class);
 				if (messageSequenceSize == null) {
-					messageSequenceSize = Integer.valueOf(0);
+					messageSequenceSize = 0;
 				}
 				return messageSequenceSize.equals(getSequenceSize())
 						&& !(this.sourceGroup != null ? this.sourceGroup.containsSequence(messageSequenceNumber)

spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingCorrelationStrategy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -21,6 +21,7 @@ import java.lang.reflect.Method;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.handler.MethodInvokingMessageProcessor;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
@@ -33,12 +34,12 @@ import org.springframework.util.Assert;
  * @author Artem Bilan
  * @author Gary Russell
  */
-public class MethodInvokingCorrelationStrategy implements CorrelationStrategy, BeanFactoryAware {
+public class MethodInvokingCorrelationStrategy implements CorrelationStrategy, BeanFactoryAware, Lifecycle {
 
 	private final MethodInvokingMessageProcessor<?> processor;
 
 	public MethodInvokingCorrelationStrategy(Object object, String methodName) {
-		this.processor = new MethodInvokingMessageProcessor<Object>(object, methodName, true);
+		this.processor = new MethodInvokingMessageProcessor<Object>(object, methodName);
 	}
 
 	public MethodInvokingCorrelationStrategy(Object object, Method method) {
@@ -60,4 +61,19 @@ public class MethodInvokingCorrelationStrategy implements CorrelationStrategy, B
 		return this.processor.processMessage(message);
 	}
 
+	@Override
+	public void start() {
+		this.processor.start();
+	}
+
+	@Override
+	public void stop() {
+		this.processor.stop();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.processor.isRunning();
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -21,6 +21,7 @@ import java.util.Collection;
 import java.util.Map;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.context.Lifecycle;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.integration.annotation.Aggregator;
 import org.springframework.integration.store.MessageGroup;
@@ -33,9 +34,12 @@ import org.springframework.messaging.Message;
  * @author Mark Fisher
  * @author Dave Syer
  * @author Gary Russell
+ * @author Artme Bilan
+ *
  * @since 2.0
  */
-public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor {
+public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor
+		implements Lifecycle {
 
 	private final MethodInvokingMessageListProcessor<Object> processor;
 
@@ -86,4 +90,19 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
 		return this.processor.process(messagesUpForProcessing, headers);
 	}
 
+	@Override
+	public void start() {
+		this.processor.start();
+	}
+
+	@Override
+	public void stop() {
+		this.processor.stop();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.processor.isRunning();
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -22,6 +22,7 @@ import java.util.Collection;
 import java.util.Map;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.util.AbstractExpressionEvaluator;
 import org.springframework.integration.util.MessagingMethodInvokerHelper;
 import org.springframework.messaging.Message;
@@ -33,7 +34,8 @@ import org.springframework.messaging.Message;
  * @author Artem Bilan
  * @since 2.0
  */
-public class MethodInvokingMessageListProcessor<T> extends AbstractExpressionEvaluator {
+public class MethodInvokingMessageListProcessor<T> extends AbstractExpressionEvaluator
+		implements Lifecycle {
 
 	private final MessagingMethodInvokerHelper<T> delegate;
 
@@ -64,6 +66,17 @@ public class MethodInvokingMessageListProcessor<T> extends AbstractExpressionEva
 		this.delegate.setBeanFactory(beanFactory);
 	}
 
+	/**
+	 * A {@code boolean} flag to use SpEL Expression evaluation or
+	 * {@link org.springframework.messaging.handler.invocation.InvocableHandlerMethod}
+	 * for target method invocation.
+	 * @param useSpelInvoker to use SpEL Expression evaluation or not.
+	 * @since 5.0
+	 */
+	public void setUseSpelInvoker(boolean useSpelInvoker) {
+		this.delegate.setUseSpelInvoker(useSpelInvoker);
+	}
+
 	public String toString() {
 		return this.delegate.toString();
 	}
@@ -80,4 +93,19 @@ public class MethodInvokingMessageListProcessor<T> extends AbstractExpressionEva
 		}
 	}
 
+	@Override
+	public void start() {
+		this.delegate.start();
+	}
+
+	@Override
+	public void stop() {
+		this.delegate.stop();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.delegate.isRunning();
+	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -20,6 +20,7 @@ import java.lang.reflect.Method;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.context.Lifecycle;
 import org.springframework.core.convert.ConversionService;
 import org.springframework.integration.store.MessageGroup;
 
@@ -28,8 +29,9 @@ import org.springframework.integration.store.MessageGroup;
  *
  * @author Marius Bogoevici
  * @author Dave Syer
+ * @author Artme Bilan
  */
-public class MethodInvokingReleaseStrategy implements ReleaseStrategy, BeanFactoryAware {
+public class MethodInvokingReleaseStrategy implements ReleaseStrategy, BeanFactoryAware, Lifecycle {
 
 	private final MethodInvokingMessageListProcessor<Boolean> adapter;
 
@@ -57,4 +59,19 @@ public class MethodInvokingReleaseStrategy implements ReleaseStrategy, BeanFacto
 		return this.adapter.process(messages.getMessages(), null);
 	}
 
+	@Override
+	public void start() {
+		this.adapter.start();
+	}
+
+	@Override
+	public void stop() {
+		this.adapter.stop();
+	}
+
+	@Override
+	public boolean isRunning() {
+		return this.adapter.isRunning();
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -21,7 +21,6 @@ import java.util.List;
 import org.aopalliance.aop.Advice;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.reactivestreams.Subscriber;
 
 import org.springframework.aop.framework.Advised;
 import org.springframework.aop.framework.ProxyFactory;
@@ -44,7 +43,6 @@ import org.springframework.integration.endpoint.ReactiveConsumer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.advice.HandleMessageAdvice;
 import org.springframework.integration.scheduling.PollerMetadata;
-import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.PollableChannel;
@@ -288,12 +286,7 @@ public class ConsumerEndpointFactoryBean
 				this.endpoint = pollingConsumer;
 			}
 			else {
-				if (this.handler instanceof Subscriber) {
-					this.endpoint = new ReactiveConsumer(channel, (Subscriber<Message<?>>) this.handler);
-				}
-				else {
-					this.endpoint = new ReactiveConsumer(channel, this.handler::handleMessage);
-				}
+				this.endpoint = new ReactiveConsumer(channel, this.handler);
 			}
 			this.endpoint.setBeanName(this.beanName);
 			this.endpoint.setBeanFactory(this.beanFactory);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -27,7 +27,6 @@ import org.aopalliance.aop.Advice;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
-import org.reactivestreams.Subscriber;
 
 import org.springframework.aop.TargetSource;
 import org.springframework.aop.framework.Advised;
@@ -65,7 +64,6 @@ import org.springframework.integration.router.AbstractMessageRouter;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.util.MessagingAnnotationUtils;
-import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.PollableChannel;
@@ -81,8 +79,6 @@ import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
-import reactor.core.publisher.DirectProcessor;
-
 /**
  * Base class for Method-level annotation post-processors.
  *
@@ -321,17 +317,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 			Assert.state(ObjectUtils.isEmpty(pollers), "A '@Poller' should not be specified for Annotation-based " +
 					"endpoint, since '" + inputChannel + "' is a SubscribableChannel (not pollable).");
 			if (inputChannel instanceof Publisher) {
-				Subscriber<Message<?>> subscriber;
-				if (handler instanceof Subscriber) {
-					subscriber = (Subscriber<Message<?>>) handler;
-				}
-				else {
-					//TODO errorConsumer, completeConsumer
-					DirectProcessor<Message<?>> directProcessor = DirectProcessor.create();
-					directProcessor.doOnNext(handler::handleMessage);
-					subscriber = directProcessor;
-				}
-				endpoint = new ReactiveConsumer(inputChannel, subscriber);
+				endpoint = new ReactiveConsumer(inputChannel, handler);
 			}
 			else {
 				endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler);

spring-integration-core/src/main/java/org/springframework/integration/endpoint/EventDrivenConsumer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -17,7 +17,6 @@
 package org.springframework.integration.endpoint;
 
 import org.springframework.context.Lifecycle;
-import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.router.MessageRouter;
 import org.springframework.integration.support.context.NamedComponent;
@@ -95,7 +94,7 @@ public class EventDrivenConsumer extends AbstractEndpoint implements Integration
 			String channelName = ((NamedComponent) this.inputChannel).getComponentName();
 			String componentType = ((NamedComponent) this.handler).getComponentType();
 			componentType = StringUtils.hasText(componentType) ? componentType : "";
-			String componentName = ((IntegrationObjectSupport) this).getComponentName();
+			String componentName = getComponentName();
 			componentName = (StringUtils.hasText(componentName) && componentName.contains("#")) ? "" : ":" + componentName;
 			StringBuffer buffer = new StringBuffer();
 			buffer.append("{" + componentType + componentName + "} as a subscriber to the '" + channelName + "' channel");

spring-integration-core/src/main/java/org/springframework/integration/endpoint/ReactiveConsumer.java

@@ -22,11 +22,13 @@ import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
+import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
 
@@ -45,13 +47,18 @@ public class ReactiveConsumer extends AbstractEndpoint {
 
 	private final Operators.SubscriberAdapter<Message<?>, Message<?>> subscriber;
 
+	private final Lifecycle lifecycleDelegate;
+
 	private volatile Publisher<Message<?>> publisher;
 
 	private ErrorHandler errorHandler;
 
-
-	public ReactiveConsumer(MessageChannel inputChannel, Consumer<Message<?>> consumer) {
-		this(inputChannel, new ConsumerSubscriber(consumer));
+	@SuppressWarnings("unchecked")
+	public ReactiveConsumer(MessageChannel inputChannel, MessageHandler messageHandler) {
+		this(inputChannel,
+				messageHandler instanceof Subscriber
+						? (Subscriber<Message<?>>) messageHandler
+						: new MessageHandlerSubscriber(messageHandler));
 	}
 
 	@SuppressWarnings("unchecked")
@@ -76,6 +83,7 @@ public class ReactiveConsumer extends AbstractEndpoint {
 			}
 
 		};
+		this.lifecycleDelegate = subscriber instanceof Lifecycle ? (Lifecycle) subscriber : null;
 	}
 
 	public void setErrorHandler(ErrorHandler errorHandler) {
@@ -93,24 +101,34 @@ public class ReactiveConsumer extends AbstractEndpoint {
 
 	@Override
 	protected void doStart() {
+		if (this.lifecycleDelegate != null) {
+			this.lifecycleDelegate.start();
+		}
 		this.publisher.subscribe(this.subscriber);
 	}
 
 	@Override
 	protected void doStop() {
 		this.subscriber.cancel();
+		if (this.lifecycleDelegate != null) {
+			this.lifecycleDelegate.stop();
+		}
 	}
 
 
-	private static final class ConsumerSubscriber implements Subscriber<Message<?>>, Receiver, Disposable, Trackable {
+	private static final class MessageHandlerSubscriber
+			implements Subscriber<Message<?>>, Receiver, Disposable, Trackable, Lifecycle {
 
 		private final Consumer<Message<?>> consumer;
 
 		private Subscription subscription;
 
-		ConsumerSubscriber(Consumer<Message<?>> consumer) {
-			Assert.notNull(consumer);
-			this.consumer = consumer;
+		private MessageHandler messageHandler;
+
+		MessageHandlerSubscriber(MessageHandler messageHandler) {
+			Assert.notNull(messageHandler, "'messageHandler' must not be null");
+			this.messageHandler = messageHandler;
+			this.consumer = this.messageHandler::handleMessage;
 		}
 
 		@Override
@@ -169,6 +187,26 @@ public class ReactiveConsumer extends AbstractEndpoint {
 			return false;
 		}
 
+
+		@Override
+		public void start() {
+			if (this.messageHandler instanceof Lifecycle) {
+				((Lifecycle) this.messageHandler).start();
+			}
+		}
+
+		@Override
+		public void stop() {
+			if (this.messageHandler instanceof Lifecycle) {
+				((Lifecycle) this.messageHandler).stop();
+			}
+		}
+
+		@Override
+		public boolean isRunning() {
+			return !(this.messageHandler instanceof Lifecycle) || ((Lifecycle) this.messageHandler).isRunning();
+		}
+
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 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.
@@ -72,6 +72,17 @@ public class MethodInvokingMessageProcessor<T> extends AbstractMessageProcessor<
 		this.delegate.setBeanFactory(beanFactory);
 	}
 
+	/**
+	 * A {@code boolean} flag to use SpEL Expression evaluation or
+	 * {@link org.springframework.messaging.handler.invocation.InvocableHandlerMethod}
+	 * for target method invocation.
+	 * @param useSpelInvoker to use SpEL Expression evaluation or not.
+	 * @since 5.0
+	 */
+	public void setUseSpelInvoker(boolean useSpelInvoker) {
+		this.delegate.setUseSpelInvoker(useSpelInvoker);
+	}
+
 	@Override
 	public void start() {
 		this.delegate.start();

spring-integration-core/src/main/java/org/springframework/integration/handler/support/CollectionArgumentResolver.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright 2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.handler.support;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.stream.Collectors;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.integration.util.AbstractExpressionEvaluator;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link HandlerMethodArgumentResolver} implementation for {@link Collection},
+ * {@link Iterator} or {@code array} {@link MethodParameter}.
+ * <p>
+ * If {@link #canProcessMessageList} is set to {@code true}, only messages
+ * with a payload of {@code Collection<Message<?>>) are supported.
+ * Depending on the {@link MethodParameter#getNestedParameterType()} the whole
+ * {@code Collection<Message<?>>} or just payloads of those messages can be use as an actual argument.
+ * <p>
+ * If the value isn't compatible with {@link MethodParameter},
+ * the {@link org.springframework.core.convert.ConversionService} is used
+ * to convert the value to the target type.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public class CollectionArgumentResolver extends AbstractExpressionEvaluator
+		implements HandlerMethodArgumentResolver {
+
+	private final boolean canProcessMessageList;
+
+	public CollectionArgumentResolver(boolean canProcessMessageList) {
+		this.canProcessMessageList = canProcessMessageList;
+	}
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		Class<?> parameterType = parameter.getParameterType();
+		return Collection.class.isAssignableFrom(parameterType)
+				|| Iterator.class.isAssignableFrom(parameterType)
+				|| parameterType.isArray();
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
+		Object value = message.getPayload();
+
+		if (this.canProcessMessageList) {
+			Assert.state(value instanceof Collection,
+					"This Argument Resolver only supports messages with a payload of Collection<Message<?>>");
+			Collection<Message<?>> messages = (Collection<Message<?>>) value;
+
+			parameter.increaseNestingLevel();
+			if (Message.class.isAssignableFrom(parameter.getNestedParameterType())) {
+				value = messages;
+			}
+			else {
+				value = messages.stream()
+						.map(Message::getPayload)
+						.collect(Collectors.toList());
+			}
+			parameter.decreaseNestingLevel();
+		}
+
+		if (Iterator.class.isAssignableFrom(parameter.getParameterType())) {
+			if (value instanceof Iterable) {
+				return ((Iterable) value).iterator();
+			}
+			else {
+				return Collections.singleton(value).iterator();
+			}
+		}
+		else {
+			return getEvaluationContext()
+					.getTypeConverter()
+					.convertValue(value,
+							TypeDescriptor.forObject(value),
+							TypeDescriptor.valueOf(parameter.getParameterType()));
+		}
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/handler/support/MapArgumentResolver.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.handler.support;
+
+import java.util.Map;
+import java.util.Properties;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.integration.util.AbstractExpressionEvaluator;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.annotation.Headers;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
+
+/**
+ * A {@link HandlerMethodArgumentResolver} implementation to resolve argument
+ * for the {@link MethodParameter} as a {@link Map} or {@link Properties}.
+ * <p>
+ * The {@link Message#getHeaders()} is used when {@link MethodParameter} is marked
+ * with the {@link Headers} annotation or {@link Message#getPayload()} isn't {@link Map}
+ * or {@link Properties} compatible.
+ * <p>
+ * If {@link MethodParameter} is of {@link Properties} type and {@link Message#getPayload()}
+ * is a {@link String} containing {@code =} symbol, the {@link MapArgumentResolver} uses
+ * {@link ConversionService} trying to convert that {@link String} to the {@link Properties} object.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public class MapArgumentResolver extends AbstractExpressionEvaluator
+		implements HandlerMethodArgumentResolver {
+
+	private static final TypeDescriptor PROPERTIES_TYPE = TypeDescriptor.valueOf(Properties.class);
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		return !parameter.hasParameterAnnotation(Payload.class)
+				&& Map.class.isAssignableFrom(parameter.getParameterType());
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
+		Object payload = message.getPayload();
+		if (Properties.class.isAssignableFrom(parameter.getParameterType())) {
+			Map<String, Object> map = message.getHeaders();
+			if (!parameter.hasParameterAnnotation(Headers.class)) {
+				if (payload instanceof Map) {
+					map = (Map<String, Object>) payload;
+				}
+				else if (payload instanceof String && ((String) payload).contains("=")) {
+					return getEvaluationContext()
+							.getTypeConverter()
+							.convertValue(payload, TypeDescriptor.valueOf(String.class), PROPERTIES_TYPE);
+				}
+			}
+			Properties properties = new Properties();
+			properties.putAll(map);
+			return properties;
+		}
+		else {
+			if (!parameter.hasParameterAnnotation(Headers.class) && payload instanceof Map) {
+				return payload;
+			}
+			else {
+				return message.getHeaders();
+			}
+		}
+
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadExpressionArgumentResolver.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.handler.support;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.expression.Expression;
+import org.springframework.integration.util.AbstractExpressionEvaluator;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
+import org.springframework.util.StringUtils;
+
+/**
+ * The {@link HandlerMethodArgumentResolver} for evaluating {@link Payload#expression()}
+ * as a SpEL expression against {@code message} and converting result to expected parameter type.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ *
+ * @see org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver
+ */
+public class PayloadExpressionArgumentResolver extends AbstractExpressionEvaluator
+		implements HandlerMethodArgumentResolver {
+
+	private final Map<MethodParameter, Expression> expressionCache = new HashMap<>();
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		Payload ann = parameter.getParameterAnnotation(Payload.class);
+		return ann != null && StringUtils.hasText(ann.expression());
+	}
+
+	@Override
+	public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
+		Expression expression = this.expressionCache.get(parameter);
+		if (expression == null) {
+			Payload ann = parameter.getParameterAnnotation(Payload.class);
+			expression = EXPRESSION_PARSER.parseExpression(ann.expression());
+			this.expressionCache.put(parameter, expression);
+		}
+		return evaluateExpression(expression, message.getPayload(), parameter.getParameterType());
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/handler/support/PayloadsArgumentResolver.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.handler.support;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.expression.Expression;
+import org.springframework.integration.annotation.Payloads;
+import org.springframework.integration.util.AbstractExpressionEvaluator;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
+import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
+
+/**
+ * The {@link HandlerMethodArgumentResolver} for resolving a {@link Collection}
+ * of {@code payloads} or expression against each {@code payload}.
+ * <p>
+ * IMPORTANT: The {@link Message} for argument resolution must contain a {@code payload}
+ * as {@link Collection} of {@link Message}s.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.0
+ */
+public class PayloadsArgumentResolver extends AbstractExpressionEvaluator
+		implements HandlerMethodArgumentResolver {
+
+	private final Map<MethodParameter, Expression> expressionCache = new HashMap<>();
+
+	@Override
+	public boolean supportsParameter(MethodParameter parameter) {
+		return parameter.hasParameterAnnotation(Payloads.class);
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception {
+		Object payload = message.getPayload();
+		Assert.state(payload instanceof Collection,
+				"This Argument Resolver support only messages with payload as Collection<Message<?>>");
+		Collection<Message<?>> messages = (Collection<Message<?>>) payload;
+
+		if (!this.expressionCache.containsKey(parameter)) {
+			Payloads payloads = parameter.getParameterAnnotation(Payloads.class);
+			String expression = payloads.value();
+			if (StringUtils.hasText(expression)) {
+				this.expressionCache.put(parameter, EXPRESSION_PARSER.parseExpression("![payload." + expression + "]"));
+			}
+			else {
+				this.expressionCache.put(parameter, null);
+			}
+
+		}
+
+		Expression expression = this.expressionCache.get(parameter);
+		if (expression != null) {
+			return evaluateExpression(expression, messages, parameter.getParameterType());
+		}
+		else {
+			List<?> payloads = messages.stream()
+					.map(Message::getPayload)
+					.collect(Collectors.toList());
+
+			return getEvaluationContext()
+					.getTypeConverter()
+					.convertValue(payloads,
+							TypeDescriptor.forObject(payloads),
+							TypeDescriptor.valueOf(parameter.getParameterType()));
+		}
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/handler/support/package-info.java

@@ -0,0 +1,4 @@
+/**
+ * Provides classes for message handlers support.
+ */
+package org.springframework.integration.handler.support;

spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessage.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2015 the original author or authors.
+ * Copyright 2014-2017 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.
@@ -21,7 +21,6 @@ import java.util.Map;
 
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
@@ -58,11 +57,6 @@ public class MutableMessage<T> implements Message<T>, Serializable {
 		this.payload = payload;
 
 		this.headers = new MutableMessageHeaders(headers);
-
-		if (headers != null) {
-			this.headers.put(MessageHeaders.ID, headers.get(MessageHeaders.ID));
-			this.headers.put(MessageHeaders.TIMESTAMP, headers.get(MessageHeaders.TIMESTAMP));
-		}
 	}
 
 

spring-integration-core/src/main/java/org/springframework/integration/support/MutableMessageHeaders.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2017 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.
@@ -17,6 +17,7 @@
 package org.springframework.integration.support;
 
 import java.util.Map;
+import java.util.UUID;
 
 import org.springframework.messaging.MessageHeaders;
 
@@ -27,6 +28,8 @@ import org.springframework.messaging.MessageHeaders;
  *
  * @author Stuart Williams
  * @author David Turanski
+ * @author Artem Bilan
+ *
  * @since 4.2
  */
 public class MutableMessageHeaders extends MessageHeaders {
@@ -34,7 +37,13 @@ public class MutableMessageHeaders extends MessageHeaders {
 	private static final long serialVersionUID = 3084692953798643018L;
 
 	public MutableMessageHeaders(Map<String, Object> headers) {
-		super(headers);
+		super(headers,
+				(headers != null ?
+						(UUID) headers.get(MessageHeaders.ID)
+						: null),
+				(headers != null ?
+						(Long) headers.get(MessageHeaders.TIMESTAMP)
+						: null));
 	}
 
 	@Override

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

@@ -34,33 +34,45 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Stream;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.aop.framework.Advised;
 import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.context.Lifecycle;
 import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ParameterNameDiscoverer;
 import org.springframework.core.annotation.AnnotationAttributes;
 import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.convert.ConversionFailedException;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.ConverterNotFoundException;
 import org.springframework.core.convert.TypeDescriptor;
 import org.springframework.expression.EvaluationException;
 import org.springframework.expression.Expression;
 import org.springframework.expression.TypeConverter;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
-import org.springframework.expression.spel.support.ReflectiveMethodResolver;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import org.springframework.integration.annotation.Payloads;
 import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.handler.support.CollectionArgumentResolver;
+import org.springframework.integration.handler.support.MapArgumentResolver;
+import org.springframework.integration.handler.support.PayloadExpressionArgumentResolver;
+import org.springframework.integration.handler.support.PayloadsArgumentResolver;
+import org.springframework.integration.support.MutableMessage;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.handler.annotation.Headers;
 import org.springframework.messaging.handler.annotation.Payload;
+import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
+import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
+import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
+import org.springframework.messaging.handler.invocation.MethodArgumentResolutionException;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
@@ -95,6 +107,9 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 
 	private static final Log logger = LogFactory.getLog(MessagingMethodInvokerHelper.class);
 
+	private final DefaultMessageHandlerMethodFactory messageHandlerMethodFactory =
+			new DefaultMessageHandlerMethodFactory();
+
 	private final Object targetObject;
 
 	private volatile String displayString;
@@ -109,7 +124,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 
 	private final HandlerMethod handlerMethod;
 
-	private final Class<?> expectedType;
+	private final TypeDescriptor expectedType;
 
 	private final boolean canProcessMessageList;
 
@@ -121,6 +136,9 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 
 	private Method method;
 
+	private boolean useSpelInvoker;
+
+
 	public MessagingMethodInvokerHelper(Object targetObject, Method method, Class<?> expectedType,
 			boolean canProcessMessageList) {
 		this(targetObject, null, method, expectedType, canProcessMessageList);
@@ -149,13 +167,34 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		this(targetObject, annotationType, (String) null, expectedType, canProcessMessageList);
 	}
 
+	/**
+	 * A {@code boolean} flag to use SpEL Expression evaluation or {@link InvocableHandlerMethod}
+	 * for target method invocation.
+	 * @param useSpelInvoker to use SpEL Expression evaluation or not.
+	 * @since 5.0
+	 */
+	public void setUseSpelInvoker(boolean useSpelInvoker) {
+		this.useSpelInvoker = useSpelInvoker;
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) {
+		super.setBeanFactory(beanFactory);
+		this.messageHandlerMethodFactory.setBeanFactory(beanFactory);
+	}
+
+	@Override
+	public void setConversionService(ConversionService conversionService) {
+		super.setConversionService(conversionService);
+		this.messageHandlerMethodFactory.setConversionService(conversionService);
+	}
 
 	public T process(Message<?> message) throws Exception {
 		ParametersWrapper parameters = new ParametersWrapper(message);
 		return processInternal(parameters);
 	}
 
-	public T process(Collection<Message<?>> messages, Map<String, ?> headers) throws Exception {
+	public T process(Collection<Message<?>> messages, Map<String, Object> headers) throws Exception {
 		ParametersWrapper parameters = new ParametersWrapper(messages, headers);
 		return processInternal(parameters);
 	}
@@ -194,16 +233,22 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		this.canProcessMessageList = canProcessMessageList;
 		Assert.notNull(method, "method must not be null");
 		this.method = method;
-		this.expectedType = expectedType;
 		this.requiresReply = expectedType != null;
 		if (expectedType != null) {
 			Assert.isTrue(method.getReturnType() != Void.class && method.getReturnType() != Void.TYPE,
 					"method must have a return type");
+			this.expectedType = TypeDescriptor.valueOf(expectedType);
 		}
+		else {
+			this.expectedType = null;
+		}
+
 		Assert.notNull(targetObject, "targetObject must not be null");
 		this.targetObject = targetObject;
 		try {
-			this.handlerMethod = new HandlerMethod(method, canProcessMessageList);
+			InvocableHandlerMethod invocableHandlerMethod =
+					this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject, method);
+			this.handlerMethod = new HandlerMethod(invocableHandlerMethod, canProcessMessageList);
 		}
 		catch (IneligibleMethodException e) {
 			throw new IllegalArgumentException(e);
@@ -220,11 +265,15 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		this.methodName = methodName;
 		this.canProcessMessageList = canProcessMessageList;
 		Assert.notNull(targetObject, "targetObject must not be null");
-		this.expectedType = expectedType;
+		if (expectedType != null) {
+			this.expectedType = TypeDescriptor.valueOf(expectedType);
+		}
+		else {
+			this.expectedType = null;
+		}
 		this.targetObject = targetObject;
-		this.requiresReply = expectedType != null;
 		Map<String, Map<Class<?>, HandlerMethod>> handlerMethodsForTarget =
-				findHandlerMethodsForTarget(targetObject, annotationType, methodName, this.requiresReply);
+				findHandlerMethodsForTarget(targetObject, annotationType, methodName, expectedType != null);
 		Map<Class<?>, HandlerMethod> handlerMethods = handlerMethodsForTarget.get(CANDIDATE_METHODS);
 		Map<Class<?>, HandlerMethod> handlerMessageMethods = handlerMethodsForTarget.get(CANDIDATE_MESSAGE_METHODS);
 		if ((handlerMethods.size() == 1 && handlerMessageMethods.isEmpty()) ||
@@ -268,39 +317,22 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 	private void prepareEvaluationContext() {
 		StandardEvaluationContext context = getEvaluationContext(false);
 		Class<?> targetType = AopUtils.getTargetClass(this.targetObject);
-
-		ReflectiveMethodResolver declaredMethodResolver = new ReflectiveMethodResolver() {
-
-			@Override
-			protected Method[] getMethods(Class<?> type) {
-				return Stream.of(type.getMethods(), type.getDeclaredMethods())
-						.flatMap(Stream::of)
-						.toArray(Method[]::new);
-			}
-
-		};
-
 		if (this.method != null) {
-			FixedMethodFilter fixedMethodFilter = new FixedMethodFilter(this.method);
-			context.registerMethodFilter(targetType, fixedMethodFilter);
-			declaredMethodResolver.registerMethodFilter(targetType, fixedMethodFilter);
+			context.registerMethodFilter(targetType, new FixedMethodFilter(this.method));
 			if (this.expectedType != null) {
 				Assert.state(context.getTypeConverter()
-								.canConvert(TypeDescriptor.valueOf((this.method).getReturnType()),
-										TypeDescriptor.valueOf(this.expectedType)),
+								.canConvert(TypeDescriptor.valueOf((this.method).getReturnType()), this.expectedType),
 						"Cannot convert to expected type (" + this.expectedType + ") from " + this.method);
 			}
 		}
 		else {
-			AnnotatedMethodFilter annotatedMethodFilter = new AnnotatedMethodFilter(this.annotationType,
-					this.methodName, this.requiresReply);
-			Assert.state(canReturnExpectedType(annotatedMethodFilter, targetType, context.getTypeConverter()),
+			AnnotatedMethodFilter filter = new AnnotatedMethodFilter(this.annotationType, this.methodName,
+					this.requiresReply);
+			Assert.state(canReturnExpectedType(filter, targetType, context.getTypeConverter()),
 					"Cannot convert to expected type (" + this.expectedType + ") from " + this.method);
-			context.registerMethodFilter(targetType, annotatedMethodFilter);
-			declaredMethodResolver.registerMethodFilter(targetType, annotatedMethodFilter);
+			context.registerMethodFilter(targetType, filter);
 		}
 		context.setVariable("target", this.targetObject);
-		context.setMethodResolvers(Collections.singletonList(declaredMethodResolver));
 	}
 
 	private boolean canReturnExpectedType(AnnotatedMethodFilter filter, Class<?> targetType,
@@ -310,39 +342,100 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		}
 		List<Method> methods = filter.filter(Arrays.asList(ReflectionUtils.getAllDeclaredMethods(targetType)));
 		for (Method method : methods) {
-			if (typeConverter.canConvert(TypeDescriptor.valueOf(method.getReturnType()),
-					TypeDescriptor.valueOf(this.expectedType))) {
+			if (typeConverter.canConvert(TypeDescriptor.valueOf(method.getReturnType()), this.expectedType)) {
 				return true;
 			}
 		}
 		return false;
 	}
 
+	@SuppressWarnings("unchecked")
 	private T processInternal(ParametersWrapper parameters) throws Exception {
 		if (!this.initialized) {
 			synchronized (this) {
 				if (!this.initialized) {
+					PayloadExpressionArgumentResolver payloadExpressionArgumentResolver =
+							new PayloadExpressionArgumentResolver();
+					payloadExpressionArgumentResolver.setBeanFactory(getBeanFactory());
+
+					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);
+					customArgumentResolvers.add(mapArgumentResolver);
+
+					this.messageHandlerMethodFactory.setCustomArgumentResolvers(customArgumentResolvers);
+					this.messageHandlerMethodFactory.afterPropertiesSet();
 					prepareEvaluationContext();
 					this.initialized = true;
 				}
 			}
 		}
 		HandlerMethod candidate = this.findHandlerMethodForParameters(parameters);
+		Expression expression = candidate.expression;
 		Assert.notNull(candidate, "No candidate methods found for messages.");
-		Expression expression = candidate.getExpression();
-		Class<?> expectedType = this.expectedType != null ? this.expectedType : candidate.method.getReturnType();
+
+		T result = null;
 		try {
-			@SuppressWarnings("unchecked")
-			T result = (T) evaluateExpression(expression, parameters, expectedType);
-			if (this.requiresReply) {
-				Assert.notNull(result,
-						"Expression evaluation result was null, but this processor requires a reply.");
+			if (this.useSpelInvoker || candidate.spelOnly) {
+				result = invokeExpression(expression, parameters);
 			}
+			else {
+				result = candidate.invoke(parameters);
+			}
+		}
+		catch (MethodArgumentResolutionException | MessageConversionException | IllegalStateException e) {
+			if (e instanceof MessageConversionException) {
+				if (e.getCause() instanceof ConversionFailedException &&
+						!(e.getCause().getCause() instanceof ConverterNotFoundException)) {
+					throw e;
+				}
+			}
+			else if (e instanceof IllegalStateException) {
+				if (e.getCause() instanceof IllegalArgumentException
+						&& !"argument type mismatch".equals(e.getCause().getMessage())) {
+					throw e;
+				}
+			}
+			if (logger.isInfoEnabled()) {
+				logger.info("Failed to invoke [ " + candidate.invocableHandlerMethod +
+						"] with provided arguments [ " + parameters + " ]. \n" +
+						"Falling back to SpEL invocation for expression [ " +
+						expression.getExpressionString() + " ]");
+			}
+
+			result = invokeExpression(expression, parameters);
+		}
+
+		if (result != null && this.expectedType != null) {
+			return (T) getEvaluationContext(true)
+					.getTypeConverter()
+					.convertValue(result, TypeDescriptor.forObject(result), this.expectedType);
+		}
+		else {
 			return result;
 		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private T invokeExpression(Expression expression, ParametersWrapper parameters) throws Exception {
+		try {
+			return (T) evaluateExpression(expression, parameters);
+		}
 		catch (Exception e) {
 			Throwable evaluationException = e;
-			if ((e instanceof EvaluationException || e instanceof MessageHandlingException) && e.getCause() != null) {
+			if ((e instanceof EvaluationException || e instanceof MessageHandlingException)
+					&& e.getCause() != null) {
 				evaluationException = e.getCause();
 			}
 			if (evaluationException instanceof Exception) {
@@ -357,14 +450,14 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 	private Map<String, Map<Class<?>, HandlerMethod>> findHandlerMethodsForTarget(final Object targetObject,
 			final Class<? extends Annotation> annotationType, final String methodName, final boolean requiresReply) {
 
-		Map<String, Map<Class<?>, HandlerMethod>> handlerMethods = new HashMap<String, Map<Class<?>, HandlerMethod>>();
+		Map<String, Map<Class<?>, HandlerMethod>> handlerMethods = new HashMap<>();
 
-		final Map<Class<?>, HandlerMethod> candidateMethods = new HashMap<Class<?>, HandlerMethod>();
-		final Map<Class<?>, HandlerMethod> candidateMessageMethods = new HashMap<Class<?>, HandlerMethod>();
-		final Map<Class<?>, HandlerMethod> fallbackMethods = new HashMap<Class<?>, HandlerMethod>();
-		final Map<Class<?>, HandlerMethod> fallbackMessageMethods = new HashMap<Class<?>, HandlerMethod>();
-		final AtomicReference<Class<?>> ambiguousFallbackType = new AtomicReference<Class<?>>();
-		final AtomicReference<Class<?>> ambiguousFallbackMessageGenericType = new AtomicReference<Class<?>>();
+		final Map<Class<?>, HandlerMethod> candidateMethods = new HashMap<>();
+		final Map<Class<?>, HandlerMethod> candidateMessageMethods = new HashMap<>();
+		final Map<Class<?>, HandlerMethod> fallbackMethods = new HashMap<>();
+		final Map<Class<?>, HandlerMethod> fallbackMessageMethods = new HashMap<>();
+		final AtomicReference<Class<?>> ambiguousFallbackType = new AtomicReference<>();
+		final AtomicReference<Class<?>> ambiguousFallbackMessageGenericType = new AtomicReference<>();
 		final Class<?> targetClass = getTargetClass(targetObject);
 		MethodFilter methodFilter = new UniqueMethodFilter(targetClass);
 		ReflectionUtils.doWithMethods(targetClass, method1 -> {
@@ -394,9 +487,11 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 					&& ObjectUtils.containsElement(new String[] { "start", "stop", "isRunning" }, method1.getName())) {
 				return;
 			}
-			HandlerMethod handlerMethod1 = null;
+			HandlerMethod handlerMethod1;
 			try {
-				handlerMethod1 = new HandlerMethod(method1, this.canProcessMessageList);
+				InvocableHandlerMethod invocableHandlerMethod =
+						this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject, method1);
+				handlerMethod1 = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList);
 			}
 			catch (IneligibleMethodException e) {
 				if (logger.isDebugEnabled()) {
@@ -476,7 +571,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 			 * Ambiguous means > 1 method that takes the same payload type, or > 1 method
 			 * that takes a Message with the same generic type.
 			 */
-			List<Method> frameworkMethods = new ArrayList<Method>();
+			List<Method> frameworkMethods = new ArrayList<>();
 			Class<?>[] allInterfaces = org.springframework.util.ClassUtils.getAllInterfacesForClass(targetClass);
 			for (Class<?> iface : allInterfaces) {
 				try {
@@ -493,8 +588,11 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 				}
 			}
 			if (frameworkMethods.size() == 1) {
-				HandlerMethod handlerMethod = new HandlerMethod(frameworkMethods.get(0), this.canProcessMessageList);
-				handlerMethods.put(CANDIDATE_METHODS, Collections.<Class<?>, HandlerMethod>singletonMap(Object.class, handlerMethod));
+				InvocableHandlerMethod invocableHandlerMethod =
+						this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject,
+								frameworkMethods.get(0));
+				HandlerMethod handlerMethod = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList);
+				handlerMethods.put(CANDIDATE_METHODS, Collections.singletonMap(Object.class, handlerMethod));
 				handlerMethods.put(CANDIDATE_MESSAGE_METHODS, candidateMessageMethods);
 				return handlerMethods;
 			}
@@ -507,7 +605,9 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		Assert.isNull(ambiguousFallbackType.get(), "Found ambiguous parameter type [" + ambiguousFallbackType
 				+ "] for method match: " + fallbackMethods.values());
 		Assert.isNull(ambiguousFallbackMessageGenericType.get(),
-				"Found ambiguous parameter type [" + ambiguousFallbackMessageGenericType + "] for method match: "
+				"Found ambiguous parameter type ["
+						+ ambiguousFallbackMessageGenericType
+						+ "] for method match: "
 						+ fallbackMethods.values());
 
 		handlerMethods.put(CANDIDATE_METHODS, fallbackMethods);
@@ -519,7 +619,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 			Map<Class<?>, HandlerMethod> candidateMessageMethods,
 			Map<Class<?>, HandlerMethod> candidateMethods) {
 		if (AopUtils.isAopProxy(targetObject)) {
-			final AtomicReference<Method> targetMethod = new AtomicReference<Method>();
+			final AtomicReference<Method> targetMethod = new AtomicReference<>();
 			Class<?>[] interfaces = ((Advised) targetObject).getProxiedInterfaces();
 			for (Class<?> clazz : interfaces) {
 				ReflectionUtils.doWithMethods(clazz, method1 -> {
@@ -533,7 +633,9 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 			}
 			Method method = targetMethod.get();
 			if (method != null) {
-				HandlerMethod handlerMethod = new HandlerMethod(method, this.canProcessMessageList);
+				InvocableHandlerMethod invocableHandlerMethod =
+						this.messageHandlerMethodFactory.createInvocableHandlerMethod(targetObject, method);
+				HandlerMethod handlerMethod = new HandlerMethod(invocableHandlerMethod, this.canProcessMessageList);
 				Class<?> targetParameterType = handlerMethod.getTargetParameterType();
 				if (handlerMethod.isMessageMethod()) {
 					if (candidateMessageMethods.containsKey(targetParameterType)) {
@@ -653,11 +755,10 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		@SuppressWarnings("unused")
 		private static final Collection<Message<?>> dummyMessages = Collections.emptyList();
 
-
-		private final Method method;
-
 		private final Expression expression;
 
+		private final InvocableHandlerMethod invocableHandlerMethod;
+
 		private final boolean canProcessMessageList;
 
 		private volatile TypeDescriptor targetParameterTypeDescriptor;
@@ -666,14 +767,25 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 
 		private volatile boolean messageMethod;
 
-		HandlerMethod(Method method, boolean canProcessMessageList) {
-			this.method = method;
+		private volatile boolean spelOnly;
+
+		HandlerMethod(InvocableHandlerMethod invocableHandlerMethod, boolean canProcessMessageList) {
+			this.invocableHandlerMethod = invocableHandlerMethod;
 			this.canProcessMessageList = canProcessMessageList;
-			this.expression = this.generateExpression(method);
+			this.expression = generateExpression(this.invocableHandlerMethod.getMethod());
 		}
 
 
-		Expression getExpression() {
+		@SuppressWarnings("unchecked")
+		public <T> T invoke(ParametersWrapper parameters) throws Exception {
+			Message<?> message = parameters.getMessage();
+			if (this.canProcessMessageList) {
+				message = new MutableMessage<>(parameters.getMessages(), parameters.getHeaders());
+			}
+			return (T) this.invocableHandlerMethod.invoke(message);
+		}
+
+		public Expression getExpression() {
 			return this.expression;
 		}
 
@@ -687,7 +799,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 
 		@Override
 		public String toString() {
-			return this.method.toString();
+			return this.invocableHandlerMethod.toString();
 		}
 
 		private Expression generateExpression(Method method) {
@@ -718,6 +830,10 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 						}
 					}
 					if (annotationType.equals(Payloads.class)) {
+						Assert.isTrue(this.canProcessMessageList,
+								"The @Payloads annotation can only be applied if method handler canProcessMessageList.");
+						Assert.isTrue(Collection.class.isAssignableFrom(parameterType),
+								"The @Payloads annotation can only be applied to a Collection-typed parameter.");
 						sb.append("messages.![payload");
 						String qualifierExpression = ((Payloads) mappingAnnotation).value();
 						if (StringUtils.hasText(qualifierExpression)) {
@@ -827,6 +943,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 				headerName = tokens[0];
 				if (StringUtils.hasText(tokens[1])) {
 					relativeExpression = "." + tokens[1];
+					this.spelOnly = true;
 				}
 			}
 			else {
@@ -844,7 +961,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 			return headerRetrievalExpression + " != null ? " + fullHeaderExpression + " : " + fallbackExpression;
 		}
 
-		private synchronized void setExclusiveTargetParameterType(TypeDescriptor targetParameterType,
+		private void setExclusiveTargetParameterType(TypeDescriptor targetParameterType,
 				MethodParameter methodParameter) {
 			if (this.targetParameterTypeDescriptor != null) {
 				throw new IneligibleMethodException("Found more than one parameter type candidate: [" +
@@ -862,24 +979,24 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		}
 	}
 
-	public class ParametersWrapper {
+	private static class ParametersWrapper {
 
 		private final Object payload;
 
 		private final Collection<Message<?>> messages;
 
-		private final Map<String, ?> headers;
+		private final Map<String, Object> headers;
 
 		private final Message<?> message;
 
-		public ParametersWrapper(Message<?> message) {
+		ParametersWrapper(Message<?> message) {
 			this.message = message;
 			this.payload = message.getPayload();
 			this.headers = message.getHeaders();
 			this.messages = null;
 		}
 
-		public ParametersWrapper(Collection<Message<?>> messages, Map<String, ?> headers) {
+		ParametersWrapper(Collection<Message<?>> messages, Map<String, Object> headers) {
 			this.payload = null;
 			this.messages = messages;
 			this.headers = headers;
@@ -887,16 +1004,18 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 		}
 
 		public Object getPayload() {
-			Assert.state(this.payload != null, "Invalid method parameter for payload: was expecting collection.");
+			Assert.state(this.payload != null,
+					"Invalid method parameter for payload: was expecting collection.");
 			return this.payload;
 		}
 
 		public Collection<Message<?>> getMessages() {
-			Assert.state(this.messages != null, "Invalid method parameter for messages: was expecting a single payload.");
+			Assert.state(this.messages != null,
+					"Invalid method parameter for messages: was expecting a single payload.");
 			return this.messages;
 		}
 
-		public Map<String, ?> getHeaders() {
+		public Map<String, Object> getHeaders() {
 			return this.headers;
 		}