GH-596 Add support for handling conversion of complex types

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java

@@ -21,6 +21,7 @@ import java.lang.reflect.GenericArrayType;
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.lang.reflect.WildcardType;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -39,6 +40,7 @@ import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import net.jodah.typetools.TypeResolver;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
 import org.apache.commons.logging.Log;
@@ -49,8 +51,10 @@ import reactor.core.publisher.Mono;
 import reactor.util.function.Tuples;
 
 
+
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.function.context.FunctionProperties;
 import org.springframework.cloud.function.context.FunctionRegistration;
 import org.springframework.cloud.function.context.FunctionRegistry;
@@ -74,6 +78,8 @@ import org.springframework.util.ObjectUtils;
 import org.springframework.util.ReflectionUtils;
 import org.springframework.util.StringUtils;
 
+
+
 /**
  *
  * Basic implementation of FunctionRegistry which maintains the cache of registered functions while
@@ -109,6 +115,9 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 
 	private List<String> declaredFunctionDefinitions;
 
+	@Autowired
+	private JsonMapper jsonMapper;
+
 	public SimpleFunctionRegistry(ConversionService conversionService, @Nullable CompositeMessageConverter messageConverter) {
 		this.conversionService = conversionService;
 		this.messageConverter = messageConverter;
@@ -791,7 +800,23 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 				}
 				if (value instanceof Message<?>) { // see AWS adapter with Optional payload
 					if (messageNeedsConversion(rawType, (Message<?>) value)) {
-						convertedValue = FunctionTypeUtils.isTypeCollection(type)
+
+						boolean convertWithHint = false;
+						Type hint = type;
+						if (FunctionTypeUtils.isTypeCollection(type)) {
+							hint = FunctionTypeUtils.getGenericType(type);
+							convertWithHint = true;
+						}
+						else if (!rawType.equals(type)) {
+//							hint = FunctionTypeUtils.getGenericType(type);
+							convertWithHint = true;
+						}
+
+
+
+
+
+						convertedValue = convertWithHint
 							? this.fromMessage((Message<?>) value, (Class<?>) rawType, FunctionTypeUtils.getGenericType(type))
 							: this.fromMessage((Message<?>) value, (Class<?>) rawType, null);
 						if (logger.isDebugEnabled()) {
@@ -812,22 +837,23 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 						}
 					}
 				}
-				else if (rawType instanceof Class<?>) { // see AWS adapter with WildardTypeImpl and Azure with Voids
-					if (this.isJson(value)) {
-						convertedValue = messageConverter
-								.fromMessage(new GenericMessage<Object>(value), (Class<?>) rawType);
-					}
-					else {
-						try {
-							convertedValue = conversionService.convert(value, (Class<?>) rawType);
-						}
-						catch (Exception e) {
-							if (value instanceof String || value instanceof byte[]) {
-								convertedValue = messageConverter
-									.fromMessage(new GenericMessage<Object>(value), (Class<?>) rawType);
-							}
-						}
-					}
+				else { //if (rawType instanceof Class<?>) { // see AWS adapter with WildardTypeImpl and Azure with Voids
+					convertedValue = this.convertNonMessageInputIfNecessary(type, value);
+//					if (this.isJson(value)) {
+//						convertedValue = messageConverter
+//								.fromMessage(new GenericMessage<Object>(value), (Class<?>) rawType);
+//					}
+//					else {
+//						try {
+//							convertedValue = conversionService.convert(value, (Class<?>) rawType);
+//						}
+//						catch (Exception e) {
+//							if (value instanceof String || value instanceof byte[]) {
+//								convertedValue = messageConverter
+//									.fromMessage(new GenericMessage<Object>(value), (Class<?>) rawType);
+//							}
+//						}
+//					}
 				}
 			}
 			if (logger.isDebugEnabled()) {
@@ -839,6 +865,35 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 			return convertedValue;
 		}
 
+		private Object convertNonMessageInputIfNecessary(Type inputType, Object input) {
+			Object convertedInput = input;
+			Class<?> rawInputType = FunctionTypeUtils.isReactive(inputType) || FunctionTypeUtils.isMessage(inputType)
+					? TypeResolver.resolveRawClass(FunctionTypeUtils.getImmediateGenericType(inputType, 0), null)
+					: TypeResolver.resolveRawClass(inputType, null);
+
+			if (JsonMapper.isJsonString(input) && !Message.class.isAssignableFrom(rawInputType)) {
+				if (FunctionTypeUtils.isMessage(inputType)) {
+					inputType = FunctionTypeUtils.getGenericType(inputType);
+				}
+				if (!(inputType instanceof WildcardType) && Object.class != inputType && SimpleFunctionRegistry.this.jsonMapper != null) {
+					try {
+						convertedInput = SimpleFunctionRegistry.this.jsonMapper.fromJson(input, inputType);
+					}
+					catch (Exception e) {
+						// ignore
+					}
+				}
+			}
+			else if (SimpleFunctionRegistry.this.conversionService != null
+					&& !rawInputType.equals(input.getClass())
+					&& SimpleFunctionRegistry.this.conversionService.canConvert(input.getClass(), rawInputType)) {
+				convertedInput = SimpleFunctionRegistry.this.conversionService.convert(input, rawInputType);
+			}
+			return convertedInput;
+		}
+
+
+
 		private Object fromMessage(Message<?> message, Class<?> rawType, Object conversionHint) {
 			Stream<?> stream = null;
 			if (message.getPayload() instanceof Collection) {

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/json/JsonMapper.java

@@ -98,6 +98,9 @@ public abstract class JsonMapper {
 	 */
 	public static boolean isJsonString(Object value) {
 		boolean isJson = false;
+		if (value instanceof byte[]) {
+			value = new String((byte[]) value, StandardCharsets.UTF_8);
+		}
 		if (value instanceof String) {
 			String str = ((String) value).trim();
 			isJson = (str.startsWith("\"") && str.endsWith("\"")) ||

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistryTests.java

@@ -483,6 +483,28 @@ public class BeanFactoryAwareFunctionRegistryTests {
 		assertThat(f.apply("Bubbles")).isEqualTo("BUBBLES");
 	}
 
+	@Test
+	public void textTypeConversionWithComplexInputType() {
+		FunctionCatalog catalog = this.configureCatalog(ComplexTypeFunctionConfiguration.class);
+		Function function = catalog.lookup("function");
+
+		// as String
+		String result = (String) function.apply("{\"key\":\"purchase\",\"data\":{\"name\":\"bike\"}}");
+		assertThat(result).isEqualTo("BIKE");
+
+		// as byte[]
+		result = (String) function.apply("{\"key\":\"purchase\",\"data\":{\"name\":\"bike\"}}".getBytes());
+		assertThat(result).isEqualTo("BIKE");
+
+		// as Message<String>
+		result = (String) function.apply(MessageBuilder.withPayload("{\"key\":\"purchase\",\"data\":{\"name\":\"bike\"}}").build());
+		assertThat(result).isEqualTo("BIKE");
+
+		// as Message<BYTE[]>
+		result = (String) function.apply(MessageBuilder.withPayload("{\"key\":\"purchase\",\"data\":{\"name\":\"bike\"}}".getBytes()).build());
+		assertThat(result).isEqualTo("BIKE");
+	}
+
 	@EnableAutoConfiguration
 	public static class PojoToMessageFunctionCompositionConfiguration {
 
@@ -919,4 +941,47 @@ public class BeanFactoryAwareFunctionRegistryTests {
 		}
 
 	}
+	@EnableAutoConfiguration
+	@Configuration
+	public static class ComplexTypeFunctionConfiguration {
+		@Bean
+		public Function<Event<String, Product>, String> function() {
+			return v -> v.getData().getName().toUpperCase();
+		}
+	}
+
+	private static class Product {
+		private String name;
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+	}
+
+	private static class Event<K, V> {
+
+		private K key;
+
+		public K getKey() {
+			return key;
+		}
+
+		public void setKey(K key) {
+			this.key = key;
+		}
+
+		private V data;
+
+		public V getData() {
+			return data;
+		}
+
+		public void setData(V data) {
+			this.data = data;
+		}
+	}
 }