GH-3052: Fix custom converters for lambdas

Fixes #3052

Starting with version `5.1`, a `LambdaMessageProcessor` is based on the
`MessageConverter` instead of plain `ConversionService`.
(A `ConfigurableCompositeMessageConverter` is used from the application
context.)
However a type conversion based on the `@IntegrationConverter` is lost
in the `GenericMessageConverter`  because it doesn't use an
`integrationConversionService` from the application context

* Change `ConfigurableCompositeMessageConverter` to configure a
 `GenericMessageConverter`  with an  `integrationConversionService`
 if any
* Fix `MessagingMethodInvokerHelper` to populate a `BeanFactory` into
its internal `ConfigurableCompositeMessageConverter`
* Ensure in the `LambdaMessageProcessorTests` that
`@IntegrationConverter` is applied for ``LambdaMessageProcessor` as well

**Cherry-pick to 5.1.x**

# Conflicts:
#	spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java
#	spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java

Author: artembilan

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

@@ -578,18 +578,20 @@ private synchronized void initialize() throws Exception {
 	 * that don't run in an application context.
 	 */
 	private void configureLocalMessageHandlerFactory() {
-		MessageConverter messageConverter = null;
+		ConfigurableCompositeMessageConverter messageConverter = null;
 		BeanFactory beanFactory = getBeanFactory();
 		if (beanFactory != null &&
 				beanFactory.containsBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)) {
 			messageConverter = beanFactory
 					.getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME,
-							MessageConverter.class);
+							ConfigurableCompositeMessageConverter.class);
 			((DefaultMessageHandlerMethodFactory) this.messageHandlerMethodFactory)
 					.setMessageConverter(messageConverter);
 		}
 		else {
 			messageConverter = new ConfigurableCompositeMessageConverter();
+			messageConverter.setBeanFactory(beanFactory);
+			messageConverter.afterPropertiesSet();
 		}
 		NullAwarePayloadArgumentResolver nullResolver = new NullAwarePayloadArgumentResolver(messageConverter);
 		PayloadExpressionArgumentResolver payloadExpressionArgumentResolver = new PayloadExpressionArgumentResolver();

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

@@ -22,8 +22,15 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.convert.ConversionService;
+import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.integration.support.json.Jackson2JsonObjectMapper;
 import org.springframework.integration.support.json.JacksonPresent;
+import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.messaging.converter.ByteArrayMessageConverter;
 import org.springframework.messaging.converter.CompositeMessageConverter;
 import org.springframework.messaging.converter.GenericMessageConverter;
@@ -47,13 +54,19 @@
  *
  * @since 5.0
  */
-public class ConfigurableCompositeMessageConverter extends CompositeMessageConverter {
+public class ConfigurableCompositeMessageConverter extends CompositeMessageConverter
+		implements BeanFactoryAware, InitializingBean {
+
+	private final boolean registerDefaults;
+
+	private BeanFactory beanFactory;
 
 	/**
 	 * Create an instance with the default converters.
 	 */
 	public ConfigurableCompositeMessageConverter() {
 		super(initDefaults());
+		this.registerDefaults = true;
 	}
 
 	/**
@@ -73,6 +86,23 @@ public ConfigurableCompositeMessageConverter(Collection<MessageConverter> conver
 		super(registerDefaults ?
 				Stream.concat(converters.stream(), initDefaults().stream()).collect(Collectors.toList())
 				: converters);
+		this.registerDefaults = registerDefaults;
+	}
+
+	@Override
+	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+		this.beanFactory = beanFactory;
+	}
+
+	@Override
+	public void afterPropertiesSet() {
+		if (this.registerDefaults) {
+			ConversionService conversionService = IntegrationUtils.getConversionService(this.beanFactory);
+			if (conversionService == null) {
+				conversionService = DefaultConversionService.getSharedInstance();
+			}
+			getConverters().add(new GenericMessageConverter(conversionService));
+		}
 	}
 
 	private static Collection<MessageConverter> initDefaults() {
@@ -90,8 +120,6 @@ private static Collection<MessageConverter> initDefaults() {
 		// TODO do we port it together with MessageConverterUtils ?
 		// converters.add(new JavaSerializationMessageConverter());
 
-		converters.add(new GenericMessageConverter());
-
 		return converters;
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java

@@ -18,23 +18,30 @@
 
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
-import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.mock;
+
+import java.util.Objects;
+import java.util.function.Function;
 
 import org.junit.Test;
+import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.BeanFactory;
-import org.springframework.integration.context.IntegrationContextUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.config.IntegrationConverter;
 import org.springframework.integration.handler.GenericHandler;
 import org.springframework.integration.handler.LambdaMessageProcessor;
-import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
 import org.springframework.integration.transformer.GenericTransformer;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.context.junit4.SpringRunner;
 
 
 /**
@@ -43,8 +50,12 @@
  *
  * @since 5.0
  */
+@RunWith(SpringRunner.class)
 public class LambdaMessageProcessorTests {
 
+	@Autowired
+	private BeanFactory beanFactory;
+
 	@Test
 	@SuppressWarnings("divzero")
 	public void testException() {
@@ -67,7 +78,7 @@ public Message<?> transform(Message<?> source) {
 			}
 
 		}, null);
-		lmp.setBeanFactory(mock(BeanFactory.class));
+		lmp.setBeanFactory(this.beanFactory);
 		GenericMessage<String> testMessage = new GenericMessage<>("foo");
 		Object result = lmp.processMessage(testMessage);
 		assertSame(testMessage, result);
@@ -77,14 +88,22 @@ public Message<?> transform(Message<?> source) {
 	public void testMessageAsArgumentLambda() {
 		LambdaMessageProcessor lmp = new LambdaMessageProcessor(
 				(GenericTransformer<Message<?>, Message<?>>) source -> messageTransformer(source), null);
-		lmp.setBeanFactory(mock(BeanFactory.class));
+		lmp.setBeanFactory(this.beanFactory);
 		GenericMessage<String> testMessage = new GenericMessage<>("foo");
 		assertThatThrownBy(() -> lmp.processMessage(testMessage)).hasCauseExactlyInstanceOf(ClassCastException.class);
 	}
 
+	@Test
+	public void testCustomConverter() {
+		LambdaMessageProcessor lmp = new LambdaMessageProcessor(Function.identity(), TestPojo.class);
+		lmp.setBeanFactory(this.beanFactory);
+		Object result = lmp.processMessage(new GenericMessage<>("foo"));
+		assertEquals(new TestPojo("foo"), result);
+	}
+
 	private void handle(GenericHandler<?> h) {
 		LambdaMessageProcessor lmp = new LambdaMessageProcessor(h, String.class);
-		lmp.setBeanFactory(getBeanFactory());
+		lmp.setBeanFactory(this.beanFactory);
 
 		lmp.processMessage(new GenericMessage<>("foo"));
 	}
@@ -94,12 +113,50 @@ private Message<?> messageTransformer(Message<?> message) {
 	}
 
 
-	private BeanFactory getBeanFactory() {
-		BeanFactory mockBeanFactory = mock(BeanFactory.class);
-		given(mockBeanFactory.getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME,
-				MessageConverter.class))
-				.willReturn(new ConfigurableCompositeMessageConverter());
-		return mockBeanFactory;
+	@Configuration
+	@EnableIntegration
+	public static class TestConfiguration {
+
+		@Bean
+		@IntegrationConverter
+		public Converter<String, TestPojo> testPojoConverter() {
+			return new Converter<String, TestPojo>() { // Cannot be lambda for explicit generic types
+
+				@Override
+				public TestPojo convert(String source) {
+					return new TestPojo(source);
+				}
+
+			};
+		}
+
+	}
+
+	private static class TestPojo {
+
+		private final String value;
+
+		TestPojo(String value) {
+			this.value = value;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o) {
+				return true;
+			}
+			if (!(o instanceof TestPojo)) {
+				return false;
+			}
+			TestPojo testPojo = (TestPojo) o;
+			return Objects.equals(this.value, testPojo.value);
+		}
+
+		@Override
+		public int hashCode() {
+			return Objects.hash(this.value);
+		}
+
 	}
 
 }