INT-4523: Add DSL convert(Class<> cls) operator (#2556)

* INT-4523: Add DSL `convert(Class<> cls)` operator

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

* Change the `LambdaMessageProcessor` to rely on the `MessageConverter`
populated by the `IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME`
This way all the Lambda-based handlers are going to work the same way
as POJO-based via `MessagingMethodInvokerHelper`
* Add `convert(Class<P> payloadType)` EIP-operator to perform similar
to POJO-based method invocation argument conversion

* * Fix `LambdaMessageProcessorTests` to inject a
`ConfigurableCompositeMessageConverter` from the mocked `BeanFactory`
* Make a `mappingJackson2MessageConverter.setStrictContentTypeMatch(true)`
* Also obtain an `ObjectMapper` from the `Jackson2JsonObjectMapper`
which is configured with the scanned possible Jackson modules.
in the `ConfigurableCompositeMessageConverter` do not try to convert
all the potential content without an appropriate JSON content-type
header

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java

@@ -587,6 +587,21 @@ public B transform(MessageProcessorSpec<?> messageProcessorSpec,
 				.transform(new MethodInvokingTransformer(processor), endpointConfigurer);
 	}
 
+	/**
+	 * Populate the {@link MessageTransformingHandler} instance
+	 * for the provided {@code payloadType} to convert at runtime.
+	 * @param payloadType the {@link Class} for expected payload type.
+	 * @param <P> the payload type - 'convert to'.
+	 * @return the current {@link IntegrationFlowDefinition}.
+	 * @since 5.1
+	 * @see MethodInvokingTransformer
+	 * @see LambdaMessageProcessor
+	 */
+	public <P> B convert(Class<P> payloadType) {
+		return transform(payloadType, p -> p);
+	}
+
+
 	/**
 	 * Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer}
 	 * for the specific {@code payloadType} to convert at runtime.
@@ -599,7 +614,7 @@ public B transform(MessageProcessorSpec<?> messageProcessorSpec,
 	 * @see LambdaMessageProcessor
 	 */
 	public <P, T> B transform(Class<P> payloadType, GenericTransformer<P, T> genericTransformer) {
-		return this.transform(payloadType, genericTransformer, null);
+		return transform(payloadType, genericTransformer, null);
 	}
 
 	/**
@@ -619,6 +634,25 @@ public <S, T> B transform(GenericTransformer<S, T> genericTransformer,
 		return this.transform(null, genericTransformer, endpointConfigurer);
 	}
 
+	/**
+	 * Populate the {@link MessageTransformingHandler} instance
+	 * for the provided {@code payloadType} to convert at runtime.
+	 * In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
+	 * @param payloadType the {@link Class} for expected payload type.
+	 * @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
+	 * @param <P> the payload type - 'transform to'.
+	 * @return the current {@link IntegrationFlowDefinition}.
+	 * @since 5.1
+	 * @see MethodInvokingTransformer
+	 * @see LambdaMessageProcessor
+	 * @see GenericEndpointSpec
+	 */
+	public <P> B convert(Class<P> payloadType,
+			Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
+
+		return transform(payloadType, p -> p, endpointConfigurer);
+	}
+
 	/**
 	 * Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer}
 	 * for the specific {@code payloadType} to convert at runtime.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -25,12 +25,10 @@
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
-import org.springframework.core.convert.ConversionService;
-import org.springframework.core.convert.TypeDescriptor;
-import org.springframework.core.convert.support.DefaultConversionService;
-import org.springframework.integration.support.utils.IntegrationUtils;
+import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.Assert;
 import org.springframework.util.ReflectionUtils;
 
@@ -48,11 +46,11 @@ public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFac
 
 	private final Method method;
 
-	private final TypeDescriptor payloadType;
+	private final Class<?> payloadType;
 
 	private final Class<?>[] parameterTypes;
 
-	private ConversionService conversionService;
+	private MessageConverter messageConverter;
 
 	public LambdaMessageProcessor(Object target, Class<?> payloadType) {
 		Assert.notNull(target, "'target' must not be null");
@@ -79,16 +77,14 @@ public LambdaMessageProcessor(Object target, Class<?> payloadType) {
 		this.method = methodValue.get();
 		this.method.setAccessible(true);
 		this.parameterTypes = this.method.getParameterTypes();
-		this.payloadType = payloadType != null ? TypeDescriptor.valueOf(payloadType) : null;
+		this.payloadType = payloadType;
 	}
 
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
-		ConversionService conversionService = IntegrationUtils.getConversionService(beanFactory);
-		if (conversionService == null) {
-			conversionService = DefaultConversionService.getSharedInstance();
-		}
-		this.conversionService = conversionService;
+		this.messageConverter =
+				beanFactory.getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME,
+						MessageConverter.class);
 	}
 
 	@Override
@@ -109,14 +105,12 @@ else if (Map.class.isAssignableFrom(parameterType)) {
 			}
 			else {
 				if (this.payloadType != null) {
-					if (Message.class.isAssignableFrom(this.payloadType.getType())) {
+					if (Message.class.isAssignableFrom(this.payloadType)) {
 						args[i] = message;
 					}
 					else {
-						args[i] = this.conversionService.convert(message.getPayload(),
-								TypeDescriptor.forObject(message.getPayload()), this.payloadType);
+						args[i] = this.messageConverter.fromMessage(message, this.payloadType);
 					}
-
 				}
 				else {
 					args[i] = message.getPayload();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 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.stream.Collectors;
 import java.util.stream.Stream;
 
+import org.springframework.integration.support.json.Jackson2JsonObjectMapper;
 import org.springframework.integration.support.json.JacksonPresent;
 import org.springframework.messaging.converter.ByteArrayMessageConverter;
 import org.springframework.messaging.converter.CompositeMessageConverter;
@@ -78,7 +79,10 @@ private static Collection<MessageConverter> initDefaults() {
 		List<MessageConverter> converters = new LinkedList<>();
 
 		if (JacksonPresent.isJackson2Present()) {
-			converters.add(new MappingJackson2MessageConverter());
+			MappingJackson2MessageConverter mappingJackson2MessageConverter = new MappingJackson2MessageConverter();
+			mappingJackson2MessageConverter.setStrictContentTypeMatch(true);
+			mappingJackson2MessageConverter.setObjectMapper(new Jackson2JsonObjectMapper().getObjectMapper());
+			converters.add(mappingJackson2MessageConverter);
 		}
 		converters.add(new ByteArrayMessageConverter());
 		converters.add(new ObjectStringMessageConverter());

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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,20 +20,25 @@
 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 org.junit.Test;
 
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.integration.context.IntegrationContextUtils;
 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;
 
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.0
  */
@@ -69,12 +74,22 @@ public Message<?> transform(Message<?> source) {
 
 	private void handle(GenericHandler<?> h) {
 		LambdaMessageProcessor lmp = new LambdaMessageProcessor(h, String.class);
-		lmp.setBeanFactory(mock(BeanFactory.class));
+		lmp.setBeanFactory(getBeanFactory());
+
 		lmp.processMessage(new GenericMessage<>("foo"));
 	}
 
 	private Message<?> messageTransformer(Message<?> message) {
 		return 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;
+	}
+
 }

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

@@ -196,7 +196,7 @@ public void testSubFlowContentEnricher() {
 	private PollableChannel codecReplyChannel;
 
 	@Test
-	public void testCodec() throws Exception {
+	public void testCodec() {
 		this.encodingFlowInput.send(new GenericMessage<>("bar"));
 		Message<?> receive = this.codecReplyChannel.receive(10000);
 		assertNotNull(receive);
@@ -246,6 +246,34 @@ public void transformWithHeader() {
 		assertNotNull(this.adviceChannel.receive(10000));
 	}
 
+	@Autowired
+	@Qualifier("convertFlow.input")
+	private MessageChannel convertFlowInput;
+
+	@Test
+	public void testConvertOperator() {
+		QueueChannel replyChannel = new QueueChannel();
+		Date date = new Date();
+		this.convertFlowInput.send(
+				MessageBuilder.withPayload("{\"name\": \"Baz\",\"date\": " + date.getTime() + "}")
+						.setHeader(MessageHeaders.CONTENT_TYPE, "application/json")
+						.setReplyChannel(replyChannel)
+						.build());
+
+		Message<?> receive = replyChannel.receive(10_000);
+
+		assertNotNull(receive);
+
+		Object payload = receive.getPayload();
+
+		assertThat(payload, instanceOf(TestPojo.class));
+
+		TestPojo testPojo = (TestPojo) payload;
+
+		assertEquals("Baz", testPojo.getName());
+		assertEquals(date, testPojo.getDate());
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {
@@ -413,6 +441,12 @@ public SomeService someService() {
 			return new SomeService();
 		}
 
+		@Bean
+		public IntegrationFlow convertFlow() {
+			return f -> f
+					.convert(TestPojo.class);
+		}
+
 	}
 
 	private static final class TestPojo {
@@ -421,6 +455,9 @@ private static final class TestPojo {
 
 		private Date date;
 
+		private TestPojo() {
+		}
+
 		private TestPojo(String name) {
 			this.name = name;
 		}
@@ -448,7 +485,7 @@ public void setDate(Date date) {
 	public static class MyCodec implements Codec {
 
 		@Override
-		public void encode(Object object, OutputStream outputStream) throws IOException {
+		public void encode(Object object, OutputStream outputStream) {
 		}
 
 		@Override
@@ -457,13 +494,13 @@ public byte[] encode(Object object) throws IOException {
 		}
 
 		@Override
-		public <T> T decode(InputStream inputStream, Class<T> type) throws IOException {
+		public <T> T decode(InputStream inputStream, Class<T> type) {
 			return null;
 		}
 
 		@SuppressWarnings("unchecked")
 		@Override
-		public <T> T decode(byte[] bytes, Class<T> type) throws IOException {
+		public <T> T decode(byte[] bytes, Class<T> type) {
 			return (T) (type.equals(String.class) ? new String(bytes) :
 					type.equals(Integer.class) ? Integer.valueOf(42) : Integer.valueOf(43));
 		}