MessageHandlingException consistency

* Use `IntegrationUtils.wrapInHandlingExceptionIfNecessary()` whenever
it is possible to avoid double wrapping into the `MessageHandlingException`
* Some code polishing for affected classes `@Nullable`, streams, diamonds etc.

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/core/MessageProducer.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.core;
 
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 
 /**
@@ -49,6 +50,7 @@ default void setOutputChannelName(String outputChannel) {
 	 * @return the channel.
 	 * @since 4.3
 	 */
+	@Nullable
 	MessageChannel getOutputChannel();
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018 the original author or authors.
+ * Copyright 2018-2019 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,9 +20,9 @@
 import org.springframework.integration.acks.AckUtils;
 import org.springframework.integration.acks.AcknowledgmentCallback;
 import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
-import org.springframework.messaging.MessageHandlingException;
 import org.springframework.util.Assert;
 
 /**
@@ -56,7 +56,9 @@ public boolean poll(MessageHandler handler) {
 			}
 			catch (Exception e) {
 				AckUtils.autoNack(ackCallback);
-				throw new MessageHandlingException(message, e);
+				throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message,
+						() -> "error occurred during handling message in 'MessageSourcePollingTemplate' ["
+								+ this + "]", e);
 			}
 			return true;
 		}

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

@@ -33,6 +33,7 @@
 import org.springframework.integration.support.management.MetricsContext;
 import org.springframework.integration.support.management.Statistics;
 import org.springframework.integration.support.management.TrackableComponent;
+import org.springframework.integration.support.management.metrics.MeterFacade;
 import org.springframework.integration.support.management.metrics.MetricsCaptor;
 import org.springframework.integration.support.management.metrics.SampleFacade;
 import org.springframework.integration.support.management.metrics.TimerFacade;
@@ -338,7 +339,7 @@ public String getManagedType() {
 
 	@Override
 	public void destroy() throws Exception {
-		this.timers.forEach(t -> t.remove());
+		this.timers.forEach(MeterFacade::remove);
 	}
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.handler;
 
 import org.springframework.integration.util.AbstractExpressionEvaluator;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 
 /**
@@ -28,6 +29,7 @@
  */
 public abstract class AbstractMessageProcessor<T> extends AbstractExpressionEvaluator implements MessageProcessor<T> {
 
+	@Nullable
 	public abstract T processMessage(Message<?> message);
 
 }

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

@@ -68,8 +68,10 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan
 
 	private boolean async;
 
+	@Nullable
 	private String outputChannelName;
 
+	@Nullable
 	private MessageChannel outputChannel;
 
 	private String[] notPropagatedHeaders;
@@ -205,6 +207,7 @@ protected void onInit() {
 	}
 
 	@Override
+	@Nullable
 	public MessageChannel getOutputChannel() {
 		if (this.outputChannelName != null) {
 			this.outputChannel = getChannelResolver().resolveDestination(this.outputChannelName);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -24,6 +24,7 @@
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.integration.handler.advice.HandleMessageAdvice;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
@@ -137,6 +138,7 @@ else if (!isAsync() && logger.isDebugEnabled()) {
 		}
 	}
 
+	@Nullable
 	protected Object doInvokeAdvisedRequestHandler(Message<?> message) {
 		return this.advisedRequestHandler.handleRequestMessage(message);
 	}
@@ -149,6 +151,7 @@ protected Object doInvokeAdvisedRequestHandler(Message<?> message) {
 	 * @param requestMessage The request message.
 	 * @return The result of handling the message, or {@code null}.
 	 */
+	@Nullable
 	protected abstract Object handleRequestMessage(Message<?> requestMessage);
 
 
@@ -166,6 +169,7 @@ protected Object doInvokeAdvisedRequestHandler(Message<?> message) {
 	 */
 	public interface RequestHandler {
 
+		@Nullable
 		Object handleRequestMessage(Message<?> requestMessage);
 
 		/**
@@ -189,6 +193,7 @@ private class AdvisedRequestHandler implements RequestHandler {
 		}
 
 		@Override
+		@Nullable
 		public Object handleRequestMessage(Message<?> requestMessage) {
 			return AbstractReplyProducingMessageHandler.this.handleRequestMessage(requestMessage);
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -16,10 +16,13 @@
 
 package org.springframework.integration.handler;
 
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 3.0
  *
  */
@@ -42,6 +45,7 @@ public void setPostProcessWithinAdvice(boolean postProcessWithinAdvice) {
 	}
 
 	@Override
+	@Nullable
 	protected final Object handleRequestMessage(Message<?> requestMessage) {
 		Object result = this.doHandleRequestMessage(requestMessage);
 		if (this.postProcessWithinAdvice || !this.hasAdviceChain()) {
@@ -51,6 +55,7 @@ protected final Object handleRequestMessage(Message<?> requestMessage) {
 	}
 
 	@Override
+	@Nullable
 	protected final Object doInvokeAdvisedRequestHandler(Message<?> message) {
 		Object result = super.doInvokeAdvisedRequestHandler(message);
 		if (!this.postProcessWithinAdvice) {
@@ -59,6 +64,7 @@ protected final Object doInvokeAdvisedRequestHandler(Message<?> message) {
 		return result;
 	}
 
+	@Nullable
 	protected abstract Object doHandleRequestMessage(Message<?> requestMessage);
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2019 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.
@@ -19,8 +19,8 @@
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
-import org.springframework.util.Assert;
 
 /**
  * An "artificial" {@link MessageProcessor} for lazy-load of target bean by its name.
@@ -29,6 +29,7 @@
  * @param <T> the expected {@link #processMessage} result type.
  *
  * @author Artem Bilan
+ *
  * @since 5.0
  */
 public class BeanNameMessageProcessor<T> implements MessageProcessor<T>, BeanFactoryAware {
@@ -48,11 +49,11 @@ public BeanNameMessageProcessor(String object, String methodName) {
 
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
-		Assert.notNull(beanFactory, "'beanFactory' must not be null");
 		this.beanFactory = beanFactory;
 	}
 
 	@Override
+	@Nullable
 	public T processMessage(Message<?> message) {
 		if (this.delegate == null) {
 			Object target = this.beanFactory.getBean(this.beanName);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2019 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.handler;
 
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 
@@ -32,6 +33,7 @@ public interface DiscardingMessageHandler extends MessageHandler {
 	 * Return the discard channel.
 	 * @return the channel.
 	 */
+	@Nullable
 	MessageChannel getDiscardChannel();
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2019 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,7 +31,9 @@
 import org.springframework.expression.MethodFilter;
 import org.springframework.expression.MethodResolver;
 import org.springframework.expression.spel.support.ReflectiveMethodResolver;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
+import org.springframework.util.CollectionUtils;
 
 /**
  * A MessageProcessor implementation that expects an Expression or expressionString
@@ -40,33 +42,40 @@
  * @author Dave Syer
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor<Object> {
 
 	public ExpressionCommandMessageProcessor() {
 	}
 
-	public ExpressionCommandMessageProcessor(MethodFilter methodFilter) {
+	public ExpressionCommandMessageProcessor(@Nullable MethodFilter methodFilter) {
 		this(methodFilter, null);
 	}
 
-	public ExpressionCommandMessageProcessor(MethodFilter methodFilter, BeanFactory beanFactory) {
+	public ExpressionCommandMessageProcessor(@Nullable MethodFilter methodFilter, @Nullable BeanFactory beanFactory) {
 		if (beanFactory != null) {
-			this.setBeanFactory(beanFactory);
+			setBeanFactory(beanFactory);
 		}
 		if (methodFilter != null) {
 			MethodResolver methodResolver = new ExpressionCommandMethodResolver(methodFilter);
-			this.getEvaluationContext(false).setMethodResolvers(Collections.singletonList(methodResolver));
+			getEvaluationContext(false).setMethodResolvers(Collections.singletonList(methodResolver));
 		}
 	}
 
+	@Override
+	public final void setBeanFactory(BeanFactory beanFactory) {
+		super.setBeanFactory(beanFactory);
+	}
 
 	/**
 	 * Evaluates the Message payload expression as a command.
 	 * @throws IllegalArgumentException if the payload is not an Exception or String
 	 */
 	@Override
+	@Nullable
 	public Object processMessage(Message<?> message) {
 		Object expression = message.getPayload();
 		if (expression instanceof Expression) {
@@ -91,18 +100,17 @@ private static final class ExpressionCommandMethodResolver extends ReflectiveMet
 
 		@Override
 		public MethodExecutor resolve(EvaluationContext context,
-				Object targetObject, String name, List<TypeDescriptor> argumentTypes) throws AccessException {
-			this.validateMethod(targetObject, name, (argumentTypes != null ? argumentTypes.size() : 0));
+				Object targetObject, String name, List<TypeDescriptor> argumentTypes)
+				throws AccessException {
+
+			validateMethod(targetObject, name, !CollectionUtils.isEmpty(argumentTypes) ? argumentTypes.size() : 0);
 			return super.resolve(context, targetObject, name, argumentTypes);
 		}
 
 		private void validateMethod(Object targetObject, String name, int argumentCount) {
-			if (this.methodFilter == null) {
-				return;
-			}
 			Class<?> type = (targetObject instanceof Class ? (Class<?>) targetObject : targetObject.getClass());
 			Method[] methods = type.getMethods();
-			List<Method> candidates = new ArrayList<Method>();
+			List<Method> candidates = new ArrayList<>();
 			for (Method method : methods) {
 				if (method.getName().equals(name) && method.getParameterTypes().length == argumentCount) {
 					candidates.add(method);
@@ -111,10 +119,12 @@ private void validateMethod(Object targetObject, String name, int argumentCount)
 			List<Method> supportedMethods = this.methodFilter.filter(candidates);
 			if (supportedMethods.size() == 0) {
 				String methodDescription = (candidates.size() > 0) ? candidates.get(0).toString() : name;
-				throw new EvaluationException("The method '" + methodDescription + "' is not supported by this command processor. " +
+				throw new EvaluationException("The method '" + methodDescription +
+						"' is not supported by this command processor. " +
 						"If using the Control Bus, consider adding @ManagedOperation or @ManagedAttribute.");
 			}
 		}
+
 	}
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -18,6 +18,7 @@
 
 import org.springframework.expression.Expression;
 import org.springframework.expression.ParseException;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
@@ -28,6 +29,7 @@
  * @author Mark Fisher
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 2.0
  */
 public class ExpressionEvaluatingMessageProcessor<T> extends AbstractMessageProcessor<T> {
@@ -51,7 +53,7 @@ public ExpressionEvaluatingMessageProcessor(Expression expression) {
 	 * @param expression The expression.
 	 * @param expectedType The expected type.
 	 */
-	public ExpressionEvaluatingMessageProcessor(Expression expression, Class<T> expectedType) {
+	public ExpressionEvaluatingMessageProcessor(Expression expression, @Nullable Class<T> expectedType) {
 		Assert.notNull(expression, "The expression must not be null");
 		try {
 			this.expression = expression;
@@ -84,7 +86,7 @@ public ExpressionEvaluatingMessageProcessor(String expression) {
 	 * @param expectedType the expected result type.
 	 * @since 5.0
 	 */
-	public ExpressionEvaluatingMessageProcessor(String expression, Class<T> expectedType) {
+	public ExpressionEvaluatingMessageProcessor(String expression, @Nullable Class<T> expectedType) {
 		try {
 			this.expression = EXPRESSION_PARSER.parseExpression(expression);
 			this.expectedType = expectedType;