Fix BeanFactory propagation for MMInvokerHelper

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

@@ -98,17 +98,17 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 
 	protected static final String SEND_TIMEOUT_ATTRIBUTE = "sendTimeout";
 
-	protected final Log logger = LogFactory.getLog(this.getClass());
+	protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
 
-	protected final List<String> messageHandlerAttributes = new ArrayList<>();
+	protected final List<String> messageHandlerAttributes = new ArrayList<>(); // NOSONAR
 
-	protected final ConfigurableListableBeanFactory beanFactory;
+	protected final ConfigurableListableBeanFactory beanFactory; // NOSONAR
 
-	protected final ConversionService conversionService;
+	protected final ConversionService conversionService; // NOSONAR
 
-	protected final DestinationResolver<MessageChannel> channelResolver;
+	protected final DestinationResolver<MessageChannel> channelResolver; // NOSONAR
 
-	protected final Class<T> annotationType;
+	protected final Class<T> annotationType; // NOSONAR
 
 	protected final Disposables disposables; // NOSONAR
 
@@ -192,7 +192,8 @@ public Object postProcess(Object bean, String beanName, Method method, List<Anno
 
 		if (AnnotatedElementUtils.isAnnotated(method, IdempotentReceiver.class.getName())
 				&& !AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
-			String[] interceptors = AnnotationUtils.getAnnotation(method, IdempotentReceiver.class).value(); // NOSONAR never null
+			String[] interceptors =
+					AnnotationUtils.getAnnotation(method, IdempotentReceiver.class).value(); // NOSONAR never null
 			for (String interceptor : interceptors) {
 				DefaultBeanFactoryPointcutAdvisor advisor = new DefaultBeanFactoryPointcutAdvisor();
 				advisor.setAdviceBeanName(interceptor);
@@ -281,9 +282,7 @@ else if (adviceChainBean instanceof Advice[]) {
 				else if (adviceChainBean instanceof Collection) {
 					@SuppressWarnings("unchecked")
 					Collection<Advice> adviceChainEntries = (Collection<Advice>) adviceChainBean;
-					for (Advice advice : adviceChainEntries) {
-						adviceChain.add(advice);
-					}
+					adviceChain.addAll(adviceChainEntries);
 				}
 				else {
 					throw new IllegalArgumentException("Invalid advice chain type:" +
@@ -346,7 +345,7 @@ protected AbstractEndpoint doCreateEndpoint(MessageHandler handler, MessageChann
 	}
 
 	protected void configurePollingEndpoint(AbstractPollingEndpoint pollingEndpoint, List<Annotation> annotations) {
-		PollerMetadata pollerMetadata = null;
+		PollerMetadata pollerMetadata;
 		Poller[] pollers = MessagingAnnotationUtils.resolveAttribute(annotations, "poller", Poller[].class);
 		if (!ObjectUtils.isEmpty(pollers)) {
 			Assert.state(pollers.length == 1,

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

@@ -56,7 +56,10 @@ public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 	public T processMessage(Message<?> message) {
 		if (this.delegate == null) {
 			Object target = this.beanFactory.getBean(this.beanName);
-			this.delegate = new MethodInvokingMessageProcessor<>(target, this.methodName);
+			MethodInvokingMessageProcessor<T> methodInvokingMessageProcessor =
+					new MethodInvokingMessageProcessor<>(target, this.methodName);
+			methodInvokingMessageProcessor.setBeanFactory(this.beanFactory);
+			this.delegate = methodInvokingMessageProcessor;
 		}
 		return this.delegate.processMessage(message);
 	}

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

@@ -437,14 +437,14 @@ private void prepareEvaluationContext() throws Exception {
 			if (this.expectedType != null) {
 				Assert.state(context.getTypeConverter()
 								.canConvert(TypeDescriptor.valueOf((this.method).getReturnType()), this.expectedType),
-						"Cannot convert to expected type (" + this.expectedType + ") from " + this.method);
+						() -> "Cannot convert to expected type (" + this.expectedType + ") from " + this.method);
 			}
 		}
 		else {
 			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);
+					() -> "Cannot convert to expected type (" + this.expectedType + ") from " + this.method);
 			context.registerMethodFilter(targetType, filter);
 		}
 		context.setVariable("target", this.targetObject);
@@ -705,9 +705,7 @@ && contentTypeIsJson(parameters.message)) {
 
 				}
 				catch (Exception e) {
-					if (logger.isDebugEnabled()) {
-						logger.debug("Failed to convert from JSON", e);
-					}
+					logger.debug("Failed to convert from JSON", e);
 				}
 			}
 		}
@@ -901,13 +899,14 @@ else if (!Modifier.isPublic(method1.getModifiers())) {
 		}
 
 		Assert.state(!fallbackMethods.isEmpty() || !fallbackMessageMethods.isEmpty(),
-				"Target object of type [" + this.targetObject.getClass() +
+				() -> "Target object of type [" + this.targetObject.getClass() +
 						"] has no eligible methods for handling Messages.");
 
-		Assert.isNull(ambiguousFallbackType.get(), "Found ambiguous parameter type [" + ambiguousFallbackType
-				+ "] for method match: " + fallbackMethods.values());
+		Assert.isNull(ambiguousFallbackType.get(),
+				() -> "Found ambiguous parameter type [" + ambiguousFallbackType +
+						"] for method match: " + fallbackMethods.values());
 		Assert.isNull(ambiguousFallbackMessageGenericType.get(),
-				"Found ambiguous parameter type ["
+				() -> "Found ambiguous parameter type ["
 						+ ambiguousFallbackMessageGenericType
 						+ "] for method match: "
 						+ fallbackMethods.values());
@@ -969,8 +968,9 @@ private void findSingleSpecifMethodOnInterfacesIfProxy(final Object targetObject
 	}
 
 	private void checkSpelInvokerRequired(final Class<?> targetClass, Method methodArg, HandlerMethod handlerMethod) {
-		UseSpelInvoker useSpel = AnnotationUtils.findAnnotation(AopUtils.getMostSpecificMethod(methodArg, targetClass),
-				UseSpelInvoker.class);
+		UseSpelInvoker useSpel =
+				AnnotationUtils.findAnnotation(AopUtils.getMostSpecificMethod(methodArg, targetClass),
+						UseSpelInvoker.class);
 		if (useSpel == null) {
 			useSpel = AnnotationUtils.findAnnotation(targetClass, UseSpelInvoker.class);
 		}
@@ -1007,14 +1007,12 @@ private Class<?> getTargetClass(Object targetObject) {
 				try {
 					// Maybe a proxy with no target - e.g. gateway
 					Class<?>[] interfaces = ((Advised) targetObject).getProxiedInterfaces();
-					if (interfaces != null && interfaces.length == 1) {
+					if (interfaces.length == 1) {
 						targetClass = interfaces[0];
 					}
 				}
 				catch (Exception e) {
-					if (logger.isDebugEnabled()) {
-						logger.debug("Exception trying to extract interface", e);
-					}
+					logger.debug("Exception trying to extract interface", e);
 				}
 			}
 		}
@@ -1135,7 +1133,10 @@ public String toString() {
 		}
 
 		private String generateExpression(Method method) {
-			StringBuilder sb = new StringBuilder("#target." + method.getName() + "(");
+			StringBuilder sb =
+					new StringBuilder("#target.")
+							.append(method.getName())
+							.append('(');
 			Class<?>[] parameterTypes = method.getParameterTypes();
 			Annotation[][] parameterAnnotations = method.getParameterAnnotations();
 			boolean hasUnqualifiedMapParameter = false;
@@ -1163,9 +1164,8 @@ private String generateExpression(Method method) {
 					}
 					if (annotationType.equals(Payloads.class)) {
 						Assert.isTrue(this.canProcessMessageList,
-								"The @Payloads annotation can only be applied if method handler " +
-										"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");
@@ -1350,9 +1350,7 @@ public static class ParametersWrapper {
 		 */
 		public static Object getHeader(Map<?, ?> headers, String header) {
 			Object object = headers.get(header);
-			if (object == null) {
-				throw new IllegalArgumentException("required header not available: " + header);
-			}
+			Assert.notNull(object, () -> "required header not available: " + header);
 			return object;
 		}
 

spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java

@@ -28,6 +28,8 @@
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.stream.Collectors;
 
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.integration.filter.ExpressionEvaluatingSelector;
@@ -97,9 +99,8 @@ public void setRecipients(List<Recipient> recipients) {
 		Assert.notEmpty(recipients, "'recipients' must not be empty");
 		Queue<Recipient> newRecipients = new ConcurrentLinkedQueue<>(recipients);
 
-		if (getBeanFactory() != null) {
-			newRecipients.forEach(recipient -> recipient.setChannelResolver(getChannelResolver()));
-		}
+		newRecipients.forEach(this::setupRecipient);
+
 		if (logger.isDebugEnabled()) {
 			logger.debug("Channel Recipients: " + this.recipients + " replaced with: " + newRecipients);
 		}
@@ -145,9 +146,7 @@ private void addRecipient(String channelName, String selectorExpression, Queue<R
 				new ExpressionEvaluatingSelector(selectorExpression);
 		expressionEvaluatingSelector.setBeanFactory(getBeanFactory());
 		Recipient recipient = new Recipient(channelName, expressionEvaluatingSelector);
-		if (getBeanFactory() != null) {
-			recipient.setChannelResolver(getChannelResolver());
-		}
+		setupRecipient(recipient);
 		recipients.add(recipient);
 	}
 
@@ -164,9 +163,7 @@ public void addRecipient(String channelName, MessageSelector selector) {
 	private void addRecipient(String channelName, MessageSelector selector, Queue<Recipient> recipients) {
 		Assert.hasText(channelName, "'channelName' must not be empty.");
 		Recipient recipient = new Recipient(channelName, selector);
-		if (getBeanFactory() != null) {
-			recipient.setChannelResolver(getChannelResolver());
-		}
+		setupRecipient(recipient);
 		recipients.add(recipient);
 	}
 
@@ -176,10 +173,18 @@ public void addRecipient(MessageChannel channel) {
 
 	public void addRecipient(MessageChannel channel, MessageSelector selector) {
 		Recipient recipient = new Recipient(channel, selector);
-		if (getBeanFactory() != null) {
+		setupRecipient(recipient);
+		this.recipients.add(recipient);
+	}
+
+	private void setupRecipient(Recipient recipient) {
+		BeanFactory beanFactory = getBeanFactory();
+		if (beanFactory != null) {
 			recipient.setChannelResolver(getChannelResolver());
+			if (recipient.selector instanceof BeanFactoryAware) {
+				((BeanFactoryAware) recipient.selector).setBeanFactory(beanFactory);
+			}
 		}
-		this.recipients.add(recipient);
 	}
 
 	@Override
@@ -225,14 +230,14 @@ public void replaceRecipients(Properties recipientMappings) {
 		for (String key : keys) {
 			Assert.notNull(key, "channelName can't be null.");
 			if (StringUtils.hasText(recipientMappings.getProperty(key))) {
-				this.addRecipient(key, recipientMappings.getProperty(key));
+				addRecipient(key, recipientMappings.getProperty(key));
 			}
 			else {
-				this.addRecipient(key);
+				addRecipient(key);
 			}
 		}
 		if (logger.isDebugEnabled()) {
-			logger.debug("Channel Recipients:" + originalRecipients	+ " replaced with:" + this.recipients);
+			logger.debug("Channel Recipients:" + originalRecipients + " replaced with:" + this.recipients);
 		}
 	}
 
@@ -259,7 +264,7 @@ protected Collection<MessageChannel> determineTargetChannels(Message<?> message)
 	@Override
 	protected void onInit() {
 		super.onInit();
-		this.recipients.forEach(recipient -> recipient.setChannelResolver(getChannelResolver()));
+		this.recipients.forEach(this::setupRecipient);
 	}
 
 	public static class Recipient {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 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,17 +17,20 @@
 package org.springframework.integration.aggregator;
 
 import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
 
 import org.junit.Before;
 import org.junit.Test;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Dave Syer
+ * @author Artem Bilan
  *
  */
 public class CorrelationStrategyAdapterTests {
@@ -43,34 +46,40 @@ public void init() {
 	public void testMethodName() {
 		MethodInvokingCorrelationStrategy adapter =
 				new MethodInvokingCorrelationStrategy(new SimpleMessageCorrelator(), "getKey");
+		adapter.setBeanFactory(mock(BeanFactory.class));
 		assertEquals("b", adapter.getCorrelationKey(message));
 	}
 
 	@Test
 	public void testCorrelationStrategyAdapterObjectMethod() {
-		MethodInvokingCorrelationStrategy adapter = new MethodInvokingCorrelationStrategy(new SimpleMessageCorrelator(),
+		MethodInvokingCorrelationStrategy adapter =
+				new MethodInvokingCorrelationStrategy(new SimpleMessageCorrelator(),
 				ReflectionUtils.findMethod(SimpleMessageCorrelator.class, "getKey", Message.class));
+		adapter.setBeanFactory(mock(BeanFactory.class));
 		assertEquals("b", adapter.getCorrelationKey(message));
 	}
 
 	@Test
 	public void testCorrelationStrategyAdapterPojoMethod() {
 		MethodInvokingCorrelationStrategy adapter =
 				new MethodInvokingCorrelationStrategy(new SimplePojoCorrelator(), "getKey");
+		adapter.setBeanFactory(mock(BeanFactory.class));
 		assertEquals("foo", adapter.getCorrelationKey(message));
 	}
 
 	@Test
 	public void testHeaderPojoMethod() {
 		MethodInvokingCorrelationStrategy adapter =
 				new MethodInvokingCorrelationStrategy(new SimpleHeaderCorrelator(), "getKey");
+		adapter.setBeanFactory(mock(BeanFactory.class));
 		assertEquals("b", adapter.getCorrelationKey(message));
 	}
 
 	@Test
 	public void testHeadersPojoMethod() {
 		MethodInvokingCorrelationStrategy adapter = new MethodInvokingCorrelationStrategy(new MultiHeaderCorrelator(),
 				ReflectionUtils.findMethod(MultiHeaderCorrelator.class, "getKey", String.class, String.class));
+		adapter.setBeanFactory(mock(BeanFactory.class));
 		assertEquals("bd", adapter.getCorrelationKey(message));
 	}