GH-474 Add TypeResolver library and simplify type discovery

For complex cases where deep hierarchies are used there was still an issue with the fix in #473.
By adding TypeResolver library we essentially simplify our discovery process

Resolves #474

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

@@ -254,6 +254,9 @@ public class BeanFactoryAwareFunctionRegistry
 	}
 
 	private Type discoverFunctionType(Object function, String... names) {
+		if (function instanceof RoutingFunction) {
+			return FunctionType.of(FunctionContextUtils.findType(applicationContext.getBeanFactory(), names)).getType();
+		}
 		boolean beanDefinitionExists = false;
 		for (int i = 0; i < names.length && !beanDefinitionExists; i++) {
 			beanDefinitionExists = this.applicationContext.getBeanFactory().containsBeanDefinition(names[i]);
@@ -267,9 +270,15 @@ public class BeanFactoryAwareFunctionRegistry
 			logger.info("BeanDefinition for function name(s) '" + Arrays.asList(names) +
 				"' can not be located. FunctionType will be based on " + function.getClass());
 		}
-		return beanDefinitionExists
-			? FunctionType.of(FunctionContextUtils.findType(applicationContext.getBeanFactory(), names)).getType()
-			: new FunctionType(function.getClass()).getType();
+
+		Type type = FunctionTypeUtils.discoverFunctionTypeFromClass(function.getClass());
+		if (beanDefinitionExists) {
+			Type t = FunctionTypeUtils.getImmediateGenericType(type, 0);
+			if (t == null || t == Object.class) {
+				type = FunctionType.of(FunctionContextUtils.findType(this.applicationContext.getBeanFactory(), names)).getType();
+			}
+		}
+		return type;
 	}
 
 	private String discoverDefaultDefinitionIfNecessary(String definition) {

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

@@ -30,6 +30,7 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
+import net.jodah.typetools.TypeResolver;
 import org.reactivestreams.Publisher;
 import reactor.util.function.Tuple2;
 
@@ -118,29 +119,18 @@ public final class FunctionTypeUtils {
 		}
 	}
 
+	@SuppressWarnings("unchecked")
 	public static Type discoverFunctionTypeFromClass(Class<?> functionalClass) {
 		Assert.isTrue(isFunctional(functionalClass), "Type must be one of Supplier, Function or Consumer");
-		Type[] generics = functionalClass.getGenericInterfaces();
-		if (ObjectUtils.isEmpty(generics)) {
-			return functionalClass;
+
+		if (Function.class.isAssignableFrom(functionalClass)) {
+			return TypeResolver.reify(Function.class, (Class<Function<?, ?>>) functionalClass);
 		}
-		else  {
-			for (Type generic : generics) {
-				if (generic instanceof ParameterizedType) {
-					Class<?> rawClsss = (Class<?>) ((ParameterizedType) generic).getRawType();
-					if (Function.class.isAssignableFrom(rawClsss)
-							|| Consumer.class.isAssignableFrom(rawClsss)
-							|| Supplier.class.isAssignableFrom(rawClsss)) {
-						return generic;
-					}
-					else {
-						return discoverFunctionTypeFromClass(rawClsss);
-					}
-				}
-				else {
-					return discoverFunctionTypeFromClass((Class<?>) generic);
-				}
-			}
+		else if (Consumer.class.isAssignableFrom(functionalClass)) {
+			return TypeResolver.reify(Consumer.class, (Class<Consumer<?>>) functionalClass);
+		}
+		else if (Supplier.class.isAssignableFrom(functionalClass)) {
+			return TypeResolver.reify(Supplier.class, (Class<Supplier<?>>) functionalClass);
 		}
 		return null;
 	}

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

@@ -41,6 +41,7 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.function.context.FunctionCatalog;
 import org.springframework.cloud.function.context.catalog.BeanFactoryAwareFunctionRegistry.FunctionInvocationWrapper;
+import org.springframework.cloud.function.context.catalog.FunctionTypeUtilsTests.ReactiveFunction;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -356,7 +357,6 @@ public class BeanFactoryAwareFunctionRegistryTests {
 		assertThat(result.getPayload()).isEqualTo("\"b2xsZWg=\"".getBytes());
 	}
 
-	@SuppressWarnings("rawtypes")
 	@Test
 	public void testMultipleValuesInOutputHandling() throws Exception {
 		FunctionCatalog catalog = this.configureCatalog(CollectionOutConfiguration.class);
@@ -421,6 +421,29 @@ public class BeanFactoryAwareFunctionRegistryTests {
 		assertThat(dateResult.getHeaders().get("accept")).isNull();
 	}
 
+	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@Test
+	public void testWithComplexHierarchyAndTypeConversion() {
+		FunctionCatalog catalog = this.configureCatalog(ReactiveFunctionImpl.class);
+		Function<Object, Flux> f = catalog.lookup("");
+		assertThat(f.apply(new GenericMessage("23")).blockFirst()).isEqualTo(23);
+		assertThat(f.apply(Flux.just("25")).blockFirst()).isEqualTo(25);
+		assertThat(f.apply(Flux.just(25)).blockFirst()).isEqualTo(25);
+	}
+
+	public interface ReactiveFunction<S, T> extends Function<Flux<S>, Flux<T>> {
+
+	}
+
+	@Component
+	@EnableAutoConfiguration
+	public static class ReactiveFunctionImpl implements ReactiveFunction<String, Integer> {
+		@Override
+		public Flux<Integer> apply(Flux<String> inFlux) {
+			return inFlux.map(v -> Integer.parseInt(v));
+		}
+	}
+
 	@SuppressWarnings("unchecked")
 	@EnableAutoConfiguration
 	public static class CollectionOutConfiguration {

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

@@ -132,33 +132,12 @@ public class FunctionTypeUtilsTests {
 	}
 
 	@Test
-	public void foo() {
+	public void testWithComplexHierarchy() {
 		FunctionType type = FunctionType.of(FunctionTypeUtils.discoverFunctionTypeFromClass(ReactiveFunctionImpl.class));
 		assertThat(String.class).isAssignableFrom(type.getInputType());
 		assertThat(Integer.class).isAssignableFrom(type.getOutputType());
 	}
 
-//	@Test
-//	public void testInputTypeByIndex() throws Exception {
-//		Type functionType = getReturnType("function");
-//		Type inputType = FunctionTypeUtils.getInputType(functionType, 0);
-//		assertThat(inputType.getTypeName()).isEqualTo(String.class.getName());
-//		inputType = FunctionTypeUtils.getInputType(functionType, 1);
-//		assertThat(inputType.getTypeName()).isEqualTo(Integer.class.getName());
-//
-//		functionType = getReturnType("multiInputOutputPublisherFunction");
-//		inputType = FunctionTypeUtils.getInputType(functionType, 0);
-//		System.out.println("Reactive: " + FunctionTypeUtils.isReactive(inputType));
-//		System.out.println("Reactive: " + FunctionTypeUtils.getPublisherType(inputType));
-//		System.out.println("Reactive: " + FunctionTypeUtils.getImmediateGenericType(inputType, 0));
-//		System.out.println(inputType);
-//
-//		functionType = getReturnType("typelessFunction");
-//		inputType = FunctionTypeUtils.getInputType(functionType, 0);
-//		System.out.println(inputType);
-//	}
-
-
 	private static Function<String, Integer> function() {
 		return null;
 	}
@@ -246,6 +225,5 @@ public class FunctionTypeUtilsTests {
 		public Flux<Integer> apply(Flux<String> inFlux) {
 			return inFlux.map(v -> Integer.parseInt(v));
 		}
-
 	}
 }