INT-4312: MessagingMethodInvokerHelper Convert

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

If there is a single eligible method, the message content is JSON
and the method parameter doesn't match the payload, perform JSON
conversion if there is an ObjectMapper available.

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

@@ -75,8 +75,11 @@
 import org.springframework.integration.handler.support.PayloadExpressionArgumentResolver;
 import org.springframework.integration.handler.support.PayloadsArgumentResolver;
 import org.springframework.integration.support.MutableMessage;
+import org.springframework.integration.support.json.JsonObjectMapper;
+import org.springframework.integration.support.json.JsonObjectMapperProvider;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.handler.annotation.Header;
@@ -160,6 +163,8 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 
 	private final Object targetObject;
 
+	private final JsonObjectMapper<?, ?> jsonObjectMapper;
+
 	private volatile String displayString;
 
 	private volatile boolean requiresReply;
@@ -192,7 +197,6 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
 
 	private BeanExpressionContext expressionContext;
 
-
 	public MessagingMethodInvokerHelper(Object targetObject, Method method, Class<?> expectedType,
 			boolean canProcessMessageList) {
 		this(targetObject, null, method, expectedType, canProcessMessageList);
@@ -253,11 +257,36 @@ public void setConversionService(ConversionService conversionService) {
 		}
 	}
 
+	@SuppressWarnings("unchecked")
 	public T process(Message<?> message) throws Exception {
-		ParametersWrapper parameters = new ParametersWrapper(message);
+		Message<?> messageToProcess = message;
+		/*
+		 * If there's a single method, the content is JSON, the payload is a
+		 * String or byte[], the parameter doesn't match the payload,
+		 * and there is a Json Object Mapper on the CP,
+		 * convert.
+		 */
+		if (this.handlerMethod != null && this.handlerMethod.getTargetParameterType() != null &&
+				this.jsonObjectMapper != null) {
+			Class<?> type = this.handlerMethod.getTargetParameterType();
+			if ((message.getPayload() instanceof String && !type.equals(String.class)
+					|| message.getPayload() instanceof byte[] && !type.equals(byte[].class))
+							&& contentTypeIsJson(message)) {
+				messageToProcess = getMessageBuilderFactory()
+						.withPayload(this.jsonObjectMapper.fromJson(message.getPayload(), type))
+						.copyHeaders(message.getHeaders())
+						.build();
+			}
+		}
+		ParametersWrapper parameters = new ParametersWrapper(messageToProcess);
 		return processInternal(parameters);
 	}
 
+	private boolean contentTypeIsJson(Message<?> message) {
+		Object contentType = message.getHeaders().get(MessageHeaders.CONTENT_TYPE);
+		return contentType != null ? contentType.toString().contains("json") : false;
+	}
+
 	public T process(Collection<Message<?>> messages, Map<String, Object> headers) throws Exception {
 		ParametersWrapper parameters = new ParametersWrapper(messages, headers);
 		return processInternal(parameters);
@@ -323,6 +352,14 @@ private MessagingMethodInvokerHelper(Object targetObject, Class<? extends Annota
 		this.handlerMessageMethods = null;
 		this.handlerMethodsList = null;
 		this.setDisplayString(targetObject, method);
+		JsonObjectMapper<?, ?> mapper;
+		try {
+			mapper = JsonObjectMapperProvider.newInstance();
+		}
+		catch (IllegalStateException e) {
+			mapper = null;
+		}
+		this.jsonObjectMapper = mapper;
 	}
 
 	private MessagingMethodInvokerHelper(Object targetObject, Class<? extends Annotation> annotationType,
@@ -365,6 +402,14 @@ private MessagingMethodInvokerHelper(Object targetObject, Class<? extends Annota
 			this.handlerMethodsList.add(this.handlerMessageMethods);
 		}
 		setDisplayString(targetObject, methodName);
+		JsonObjectMapper<?, ?> mapper;
+		try {
+			mapper = JsonObjectMapperProvider.newInstance();
+		}
+		catch (IllegalStateException e) {
+			mapper = null;
+		}
+		this.jsonObjectMapper = mapper;
 	}
 
 	private void setDisplayString(Object targetObject, Object targetMethod) {

spring-integration-core/src/test/java/org/springframework/integration/handler/MethodInvokingMessageProcessorTests.java

@@ -31,11 +31,13 @@
 import static org.mockito.Mockito.mock;
 
 import java.lang.reflect.Method;
+import java.util.Collections;
 import java.util.Date;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -66,6 +68,7 @@
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.handler.annotation.Header;
 import org.springframework.messaging.support.GenericMessage;
@@ -867,6 +870,18 @@ public void testUseSpelInvoker() throws Exception {
 				TestUtils.getPropertyValue(helper, "handlerMethod.expression.configuration.compilerMode"));
 	}
 
+	@Test
+	public void testSingleMethodJson() throws Exception {
+		SingleMethodJsonWithSpELBean bean = new SingleMethodJsonWithSpELBean();
+		MessagingMethodInvokerHelper<?> helper = new MessagingMethodInvokerHelper<>(bean,
+				SingleMethodJsonWithSpELBean.class.getDeclaredMethod("foo", SingleMethodJsonWithSpELBean.Foo.class),
+				false);
+		Message<?> message = new GenericMessage<>("{\"bar\":\"bar\"}",
+				Collections.singletonMap(MessageHeaders.CONTENT_TYPE, "application/json"));
+		helper.process(message);
+		assertThat(bean.foo.bar, equalTo("bar"));
+	}
+
 	private DirectFieldAccessor compileImmediate(MethodInvokingMessageProcessor processor) {
 		// Update the parser configuration compiler mode
 		SpelParserConfiguration config = TestUtils.getPropertyValue(processor,
@@ -1147,6 +1162,40 @@ public void buz(String buz) {
 
 	}
 
+	public static class SingleMethodJsonWithSpELBean {
+
+		private Foo foo;
+
+		private final CountDownLatch latch = new CountDownLatch(1);
+
+		@ServiceActivator(inputChannel = "foo")
+		@UseSpelInvoker
+		public void foo(Foo foo) {
+			this.foo = foo;
+			this.latch.countDown();
+		}
+
+		public static class Foo {
+
+			private String bar;
+
+			public String getBar() {
+				return this.bar;
+			}
+
+			public void setBar(String bar) {
+				this.bar = bar;
+			}
+
+			@Override
+			public String toString() {
+				return "Foo [bar=" + this.bar + "]";
+			}
+
+		}
+
+	}
+
 	/*
 	 * Public for SpEL access.
 	 */