Add "wrapper" type methods to FunctionInspector

These can be used to more reliably discover whether the user
has declared a function with flux types or "bare" POJOs. They
then pave the way to supporting single valued types in a special
way.

Also consolidate and simplify the logic in FunctionInspector

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

@@ -17,14 +17,17 @@
 package org.springframework.cloud.function.context;
 
 import java.lang.reflect.Field;
+import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Consumer;
@@ -37,6 +40,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.config.ConstructorArgumentValues.ValueHolder;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
@@ -55,6 +59,7 @@ import org.springframework.core.convert.ConversionService;
 import org.springframework.core.convert.support.DefaultConversionService;
 import org.springframework.core.io.FileSystemResource;
 import org.springframework.core.type.StandardMethodMetadata;
+import org.springframework.core.type.classreading.MethodMetadataReadingVisitor;
 import org.springframework.stereotype.Component;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.ReflectionUtils;
@@ -101,6 +106,16 @@ public class ContextFunctionCatalogAutoConfiguration {
 			this.processor = processor;
 		}
 
+		@Override
+		public Class<?> getInputWrapper(String name) {
+			return processor.findInputWrapper(name);
+		}
+
+		@Override
+		public Class<?> getOutputWrapper(String name) {
+			return processor.findOutputWrapper(name);
+		}
+
 		@Override
 		public Class<?> getInputType(String name) {
 			return processor.findInputType(name);
@@ -320,47 +335,23 @@ public class ContextFunctionCatalogAutoConfiguration {
 		}
 
 		private boolean isFluxFunction(String name, Function<?, ?> function) {
-			Boolean fluxTypes = this.hasFluxTypes(name, 2);
+			boolean fluxTypes = this.hasFluxTypes(name);
-			return (fluxTypes != null) ? fluxTypes
+			return fluxTypes || FunctionUtils.isFluxFunction(function);
-					: FunctionUtils.isFluxFunction(function);
 		}
 
 		private boolean isFluxConsumer(String name, Consumer<?> consumer) {
-			Boolean fluxTypes = this.hasFluxTypes(name, 1);
+			boolean fluxTypes = this.hasFluxTypes(name);
-			return (fluxTypes != null) ? fluxTypes
+			return fluxTypes || FunctionUtils.isFluxConsumer(consumer);
-					: FunctionUtils.isFluxConsumer(consumer);
 		}
 
 		private boolean isFluxSupplier(String name, Supplier<?> supplier) {
-			Boolean fluxTypes = this.hasFluxTypes(name, 1);
+			boolean fluxTypes = this.hasFluxTypes(name);
-			return (fluxTypes != null) ? fluxTypes
+			return fluxTypes || FunctionUtils.isFluxSupplier(supplier);
-					: FunctionUtils.isFluxSupplier(supplier);
 		}
 
-		private Boolean hasFluxTypes(String name, int numTypes) {
+		private boolean hasFluxTypes(String name) {
-			if (this.registry.containsBeanDefinition(name)) {
+			return FunctionInspector.isWrapper(findInputWrapper(name))
-				BeanDefinition beanDefinition = this.registry.getBeanDefinition(name);
+					|| FunctionInspector.isWrapper(findOutputWrapper(name));
-				Object source = beanDefinition.getSource();
-				if (source instanceof StandardMethodMetadata) {
-					StandardMethodMetadata metadata = (StandardMethodMetadata) source;
-					Type returnType = metadata.getIntrospectedMethod()
-							.getGenericReturnType();
-					if (returnType instanceof ParameterizedType) {
-						Type[] types = ((ParameterizedType) returnType)
-								.getActualTypeArguments();
-						if (types != null && types.length == numTypes) {
-							String fluxClassName = Flux.class.getName();
-							for (Type t : types) {
-								if (!(t.getTypeName().startsWith(fluxClassName))) {
-									return false;
-								}
-							}
-							return true;
-						}
-					}
-				}
-			}
-			return null;
 		}
 
 		private boolean isGenericSupplier(ConfigurableListableBeanFactory factory,
@@ -396,53 +387,34 @@ public class ContextFunctionCatalogAutoConfiguration {
 											String.class)));
 		}
 
-		private Class<?> findType(AbstractBeanDefinition definition,
+		private Class<?> findType(String name, AbstractBeanDefinition definition,
 				ParamType paramType) {
 			Object source = definition.getSource();
 			Type param;
 			// Start by assuming output -> Function
-			int index = paramType == ParamType.OUTPUT ? 1 : 0;
+			int index = paramType.isOutput() ? 1 : 0;
 			if (source instanceof StandardMethodMetadata) {
-				ParameterizedType type;
+				// Standard @Bean metadata
-				type = (ParameterizedType) ((StandardMethodMetadata) source)
+				ParameterizedType type = (ParameterizedType) ((StandardMethodMetadata) source)
 						.getIntrospectedMethod().getGenericReturnType();
-				if (type.getActualTypeArguments().length == 1) {
+				param = extractType(type, paramType, index);
-					// There's only one
-					index = 0;
-				}
-				Type typeArgumentAtIndex = type.getActualTypeArguments()[index];
-				if (typeArgumentAtIndex instanceof ParameterizedType && Flux.class
-						.equals(((ParameterizedType) typeArgumentAtIndex).getRawType())) {
-					param = ((ParameterizedType) typeArgumentAtIndex)
-							.getActualTypeArguments()[0];
-				}
-				else {
-					param = typeArgumentAtIndex;
-				}
 			}
 			else if (source instanceof FileSystemResource) {
 				try {
 					Type type = ClassUtils.forName(definition.getBeanClassName(), null);
-					if (type instanceof ParameterizedType) {
+					param = extractType(type, paramType, index);
-						Type typeArgumentAtIndex = ((ParameterizedType) type)
-								.getActualTypeArguments()[index];
-						if (typeArgumentAtIndex instanceof ParameterizedType) {
-							param = ((ParameterizedType) typeArgumentAtIndex)
-									.getActualTypeArguments()[0];
-						}
-						else {
-							param = typeArgumentAtIndex;
-						}
-					}
-					else {
-						param = type;
-					}
 				}
 				catch (ClassNotFoundException e) {
 					throw new IllegalStateException(
 							"Cannot instrospect bean: " + definition, e);
 				}
 			}
+			else if (source instanceof MethodMetadataReadingVisitor) {
+				// A component scan with @Beans
+				MethodMetadataReadingVisitor visitor = (MethodMetadataReadingVisitor) source;
+				Type type = findBeanType(definition, visitor);
+				param = extractType(type, paramType, index);
+			}
 			else {
 				ResolvableType resolvable = (ResolvableType) getField(definition,
 						"targetType");
@@ -458,8 +430,49 @@ public class ContextFunctionCatalogAutoConfiguration {
 				ParameterizedType concrete = (ParameterizedType) param;
 				param = concrete.getRawType();
 			}
-			return ClassUtils.resolveClassName(param.getTypeName(),
+			return (Class<?>) param;
-					registry.getClass().getClassLoader());
+		}
+
+		private Type findBeanType(AbstractBeanDefinition definition,
+				MethodMetadataReadingVisitor visitor) {
+			Class<?> factory = ClassUtils
+					.resolveClassName(visitor.getDeclaringClassName(), null);
+			List<Class<?>> params = new ArrayList<>();
+			for (ValueHolder holder : definition.getConstructorArgumentValues()
+					.getIndexedArgumentValues().values()) {
+				params.add(ClassUtils.resolveClassName(holder.getType(), null));
+			}
+			Method method = ReflectionUtils.findMethod(factory, visitor.getMethodName(),
+					params.toArray(new Class<?>[0]));
+			Type type = method.getGenericReturnType();
+			return type;
+		}
+
+		private Type extractType(Type type, ParamType paramType, int index) {
+			Type param;
+			if (type instanceof ParameterizedType) {
+				ParameterizedType parameterizedType = (ParameterizedType) type;
+				if (parameterizedType.getActualTypeArguments().length == 1) {
+					// There's only one
+					index = 0;
+				}
+				Type typeArgumentAtIndex = parameterizedType
+						.getActualTypeArguments()[index];
+				if (typeArgumentAtIndex instanceof ParameterizedType
+						&& FunctionInspector.isWrapper(
+								((ParameterizedType) typeArgumentAtIndex).getRawType())
+						&& !paramType.isWrapper()) {
+					param = ((ParameterizedType) typeArgumentAtIndex)
+							.getActualTypeArguments()[0];
+				}
+				else {
+					param = typeArgumentAtIndex;
+				}
+			}
+			else {
+				param = type;
+			}
+			return param;
 		}
 
 		private Object getField(Object target, String name) {
@@ -468,11 +481,30 @@ public class ContextFunctionCatalogAutoConfiguration {
 			return ReflectionUtils.getField(field, target);
 		}
 
+		private Class<?> findInputWrapper(String name) {
+			if (!registry.containsBeanDefinition(name)) {
+				return Object.class;
+			}
+			return findType(name,
+					(AbstractBeanDefinition) registry.getBeanDefinition(name),
+					ParamType.INPUT_WRAPPER);
+		}
+
+		private Class<?> findOutputWrapper(String name) {
+			if (!registry.containsBeanDefinition(name)) {
+				return Object.class;
+			}
+			return findType(name,
+					(AbstractBeanDefinition) registry.getBeanDefinition(name),
+					ParamType.OUTPUT_WRAPPER);
+		}
+
 		private Class<?> findInputType(String name) {
 			if (!registry.containsBeanDefinition(name)) {
 				return Object.class;
 			}
-			return findType((AbstractBeanDefinition) registry.getBeanDefinition(name),
+			return findType(name,
+					(AbstractBeanDefinition) registry.getBeanDefinition(name),
 					ParamType.INPUT);
 		}
 
@@ -481,11 +513,23 @@ public class ContextFunctionCatalogAutoConfiguration {
 				return Object.class;
 			}
 			BeanDefinition definition = registry.getBeanDefinition(name);
-			return findType((AbstractBeanDefinition) definition, ParamType.OUTPUT);
+			return findType(name, (AbstractBeanDefinition) definition, ParamType.OUTPUT);
 		}
 
 		static enum ParamType {
-			INPUT, OUTPUT;
+			INPUT, OUTPUT, INPUT_WRAPPER, OUTPUT_WRAPPER;
+
+			public boolean isOutput() {
+				return this == OUTPUT || this == OUTPUT_WRAPPER;
+			}
+
+			public boolean isInput() {
+				return this == INPUT || this == INPUT_WRAPPER;
+			}
+
+			public boolean isWrapper() {
+				return this == OUTPUT_WRAPPER || this == INPUT_WRAPPER;
+			}
 		}
 	}
 }

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

@@ -16,6 +16,12 @@
 
 package org.springframework.cloud.function.context;
 
+import java.lang.reflect.Type;
+import java.util.Optional;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 /**
  * @author Dave Syer
  *
@@ -25,9 +31,19 @@ public interface FunctionInspector {
 	Class<?> getInputType(String name);
 
 	Class<?> getOutputType(String name);
-	
+
+	Class<?> getInputWrapper(String name);
+
+	Class<?> getOutputWrapper(String name);
+
 	Object convert(String name, String value);
-	
+
 	String getName(Object function);
 
+	// Maybe make this a default method?
+	static boolean isWrapper(Type type) {
+		return Flux.class.equals(type) || Mono.class.equals(type)
+				|| Optional.class.equals(type);
+	}
+
 }

spring-cloud-function-context/test/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfigurationTests.java

@@ -30,9 +30,12 @@ import org.junit.Test;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.function.test.GenericFunction;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -68,6 +71,34 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 		assertThat(context.getBean("function")).isInstanceOf(Function.class);
 		assertThat(catalog.lookupFunction("function")).isInstanceOf(Function.class);
 		assertThat(inspector.getInputType("function")).isAssignableFrom(Map.class);
+		assertThat(inspector.getInputWrapper("function")).isAssignableFrom(Map.class);
+	}
+
+	@Test
+	public void genericFluxFunction() {
+		create(GenericFluxConfiguration.class);
+		assertThat(context.getBean("function")).isInstanceOf(Function.class);
+		assertThat(catalog.lookupFunction("function")).isInstanceOf(Function.class);
+		assertThat(inspector.getInputType("function")).isAssignableFrom(Map.class);
+		assertThat(inspector.getInputWrapper("function")).isAssignableFrom(Flux.class);
+	}
+
+	@Test
+	public void externalFunction() {
+		create(ExternalConfiguration.class);
+		assertThat(context.getBean("function")).isInstanceOf(Function.class);
+		assertThat(catalog.lookupFunction("function")).isInstanceOf(Function.class);
+		assertThat(inspector.getInputType("function")).isAssignableFrom(Map.class);
+		assertThat(inspector.getInputWrapper("function")).isAssignableFrom(Map.class);
+	}
+
+	@Test
+	public void componentScanFunction() {
+		create(ComponentScanConfiguration.class);
+		assertThat(context.getBean("function")).isInstanceOf(Function.class);
+		assertThat(catalog.lookupFunction("function")).isInstanceOf(Function.class);
+		assertThat(inspector.getInputType("function")).isAssignableFrom(Map.class);
+		assertThat(inspector.getInputWrapper("function")).isAssignableFrom(Map.class);
 	}
 
 	@Test
@@ -149,6 +180,28 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 		}
 	}
 
+	@EnableAutoConfiguration
+	@Configuration
+	@Import(GenericFunction.class)
+	protected static class ExternalConfiguration {
+	}
+
+	@EnableAutoConfiguration
+	@Configuration
+	@ComponentScan(basePackageClasses=GenericFunction.class)
+	protected static class ComponentScanConfiguration {
+	}
+
+	@EnableAutoConfiguration
+	@Configuration
+	protected static class GenericFluxConfiguration {
+		@Bean
+		public Function<Flux<Map<String, String>>, Flux<Map<String, String>>> function() {
+			return flux -> flux.map(m -> m.entrySet().stream().collect(Collectors
+					.toMap(e -> e.getKey(), e -> e.getValue().toString().toUpperCase())));
+		}
+	}
+
 	@EnableAutoConfiguration
 	@Configuration
 	protected static class QualifiedConfiguration {
@@ -184,3 +237,4 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 	}
 
 }
+

spring-cloud-function-context/test/java/org/springframework/cloud/function/test/GenericFunction.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2016-2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.function.test;
+
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@Configuration
+public class GenericFunction {
+	@Bean
+	public Function<Map<String, String>, Map<String, String>> function() {
+		return m -> m.entrySet().stream().collect(Collectors.toMap(e -> e.getKey(),
+				e -> e.getValue().toString().toUpperCase()));
+	}
+}
+

spring-cloud-function-core/src/main/java/org/springframework/cloud/function/support/FunctionUtils.java

@@ -60,7 +60,10 @@ public abstract class FunctionUtils {
 		}
 		String[] types = getParameterizedTypeNames(supplier, Supplier.class);
 		if (ObjectUtils.isEmpty(types)) {
-			return true;
+			// Assume if there is no generic information then the function is not
+			// expecting a flux. N.B. this isn't very accurate. It is better to use a
+			// FunctionInspector if one is available.
+			return false;
 		}
 		return (types[0].startsWith(FLUX_CLASS_NAME));
 	}

spring-cloud-function-deployer/src/main/java/org/springframework/cloud/function/deployer/FunctionExtractingFunctionCatalog.java

@@ -75,6 +75,16 @@ public class FunctionExtractingFunctionCatalog implements FunctionCatalog, Funct
 		return (Class<?>) inspect(name, "getOutputType");
 	}
 
+	@Override
+	public Class<?> getInputWrapper(String name) {
+		return (Class<?>) inspect(name, "getInputWrapper");
+	}
+
+	@Override
+	public Class<?> getOutputWrapper(String name) {
+		return (Class<?>) inspect(name, "getOutputWrapper");
+	}
+
 	@Override
 	public Object convert(String name, String value) {
 		return inspect(name, "convert");

spring-cloud-function-samples/spring-cloud-function-sample-pojo/pom.xml

@@ -20,7 +20,7 @@
 	<properties>
 		<java.version>1.8</java.version>
 		<spring-cloud-function.version>1.0.0.BUILD-SNAPSHOT</spring-cloud-function.version>
-		<wrapper.version>1.0.1.RELEASE</wrapper.version>
+		<wrapper.version>1.0.2.RELEASE</wrapper.version>
 		<reactor.version>3.0.7.BUILD-SNAPSHOT</reactor.version>
 	</properties>
 

spring-cloud-function-samples/spring-cloud-function-sample-pojo/src/main/java/com/example/LowercaseConfiguration.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2016-2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.example;
+
+import java.util.function.Function;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ *
+ */
+@Configuration
+public class LowercaseConfiguration {
+
+	@Bean
+	public Function<Flux<Foo>, Flux<Bar>> lowercase() {
+		return flux -> flux.log().map(value -> new Bar(value.lowercase()));
+	}
+
+}

spring-cloud-function-samples/spring-cloud-function-sample-pojo/src/main/java/com/example/SampleApplication.java

@@ -37,11 +37,6 @@ public class SampleApplication {
 		return () -> Flux.fromArray(new Bar[] { new Bar("foo"), new Bar("bar") }).log();
 	}
 
-	@Bean
-	public Function<Flux<Foo>, Flux<Bar>> lowercase() {
-		return flux -> flux.log().map(value -> new Bar(value.lowercase()));
-	}
-
 	public static void main(String[] args) throws Exception {
 		SpringApplication.run(SampleApplication.class, args);
 	}

spring-cloud-function-samples/spring-cloud-function-sample-pojo/src/test/java/com/example/SampleApplicationTests.java

@@ -52,4 +52,11 @@ public class SampleApplicationTests {
 				String.class)).isEqualTo("[{\"value\":\"FOO\"}]");
 	}
 
+	@Test
+	public void lowercase() {
+		assertThat(new TestRestTemplate().postForObject(
+				"http://localhost:" + port + "/lowercase", "[{\"value\":\"Foo\"}]",
+				String.class)).isEqualTo("[{\"value\":\"foo\"}]");
+	}
+
 }

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/FunctionController.java

@@ -25,8 +25,6 @@ import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.cloud.function.context.FunctionInspector;
-import org.springframework.cloud.function.support.FluxSupplier;
-import org.springframework.cloud.function.support.FunctionUtils;
 import org.springframework.cloud.function.web.flux.request.FluxRequest;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
@@ -95,11 +93,7 @@ public class FunctionController {
 		return supplier(supplier);
 	}
 
-	@SuppressWarnings({ "unchecked", "rawtypes" })
 	private Flux<?> supplier(Supplier<Flux<?>> supplier) {
-		if (!FunctionUtils.isFluxSupplier(supplier)) {
-			supplier = new FluxSupplier(supplier);
-		}
 		Flux<?> result = supplier.get();
 		if (logger.isDebugEnabled()) {
 			logger.debug("Handled GET with supplier");