Fix and test bug with generic input or output types

Without this fix a Function<Foo<Bar>,...> shows as having an
input type of Bar - we need to only take the parameter if the
raw type is Flux.

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

@@ -396,21 +396,23 @@ public class ContextFunctionCatalogAutoConfiguration {
 											String.class)));
 		}
 
-		private Class<?> findType(AbstractBeanDefinition definition, ParamType paramType) {
+		private Class<?> findType(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 == ParamType.OUTPUT ? 1 : 0;
 			if (source instanceof StandardMethodMetadata) {
 				ParameterizedType type;
 				type = (ParameterizedType) ((StandardMethodMetadata) source)
 						.getIntrospectedMethod().getGenericReturnType();
-				if (type.getActualTypeArguments().length==1) {
+				if (type.getActualTypeArguments().length == 1) {
 					// There's only one
 					index = 0;
 				}
 				Type typeArgumentAtIndex = type.getActualTypeArguments()[index];
-				if (typeArgumentAtIndex instanceof ParameterizedType) {
+				if (typeArgumentAtIndex instanceof ParameterizedType && Flux.class
+						.equals(((ParameterizedType) typeArgumentAtIndex).getRawType())) {
 					param = ((ParameterizedType) typeArgumentAtIndex)
 							.getActualTypeArguments()[0];
 				}
@@ -470,17 +472,18 @@ public class ContextFunctionCatalogAutoConfiguration {
 			if (!registry.containsBeanDefinition(name)) {
 				return Object.class;
 			}
-			return findType((AbstractBeanDefinition) registry.getBeanDefinition(name), ParamType.INPUT);
+			return findType((AbstractBeanDefinition) registry.getBeanDefinition(name),
+					ParamType.INPUT);
 		}
 
 		private Class<?> findOutputType(String name) {
-			if (name==null || !registry.containsBeanDefinition(name)) {
+			if (name == null || !registry.containsBeanDefinition(name)) {
 				return Object.class;
 			}
 			BeanDefinition definition = registry.getBeanDefinition(name);
 			return findType((AbstractBeanDefinition) definition, ParamType.OUTPUT);
 		}
-		
+
 		static enum ParamType {
 			INPUT, OUTPUT;
 		}

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

@@ -18,9 +18,11 @@ package org.springframework.cloud.function.context;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import org.junit.After;
 import org.junit.Test;
@@ -44,6 +46,7 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 
 	private ConfigurableApplicationContext context;
 	private InMemoryFunctionCatalog catalog;
+	private FunctionInspector inspector;
 
 	@After
 	public void close() {
@@ -59,6 +62,14 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 		assertThat(catalog.lookupFunction("function")).isInstanceOf(Function.class);
 	}
 
+	@Test
+	public void genericFunction() {
+		create(GenericConfiguration.class);
+		assertThat(context.getBean("function")).isInstanceOf(Function.class);
+		assertThat(catalog.lookupFunction("function")).isInstanceOf(Function.class);
+		assertThat(inspector.getInputType("function")).isAssignableFrom(Map.class);
+	}
+
 	@Test
 	public void simpleSupplier() {
 		create(SimpleConfiguration.class);
@@ -104,26 +115,40 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 	private void create(Class<?>... types) {
 		context = new SpringApplicationBuilder((Object[]) types).run();
 		catalog = context.getBean(InMemoryFunctionCatalog.class);
+		inspector = context.getBean(FunctionInspector.class);
 	}
 
 	@EnableAutoConfiguration
 	@Configuration
 	protected static class SimpleConfiguration {
 		private List<String> list = new ArrayList<>();
+
 		@Bean
 		public Function<String, String> function() {
 			return value -> value.toUpperCase();
 		}
+
 		@Bean
 		public Supplier<String> supplier() {
 			return () -> "hello";
 		}
+
 		@Bean
 		public Consumer<String> consumer() {
 			return value -> list.add(value);
 		}
 	}
 
+	@EnableAutoConfiguration
+	@Configuration
+	protected static class GenericConfiguration {
+		@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()));
+		}
+	}
+
 	@EnableAutoConfiguration
 	@Configuration
 	protected static class QualifiedConfiguration {
@@ -137,7 +162,7 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 	@EnableAutoConfiguration
 	@Configuration
 	protected static class AliasConfiguration {
-		@Bean({"function", "other"})
+		@Bean({ "function", "other" })
 		public Function<String, String> function() {
 			return value -> value.toUpperCase();
 		}
@@ -148,8 +173,10 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 	protected static class RegistrationConfiguration {
 		@Bean
 		public FunctionRegistration<Function<String, String>> registration() {
-			return new FunctionRegistration<Function<String, String>>(function()).name("other");
+			return new FunctionRegistration<Function<String, String>>(function())
+					.name("other");
 		}
+
 		@Bean
 		public Function<String, String> function() {
 			return value -> value.toUpperCase();

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

@@ -26,6 +26,9 @@ import javax.servlet.http.HttpServletRequest;
 import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.cloud.function.context.FunctionInspector;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.Ordered;
@@ -46,6 +49,9 @@ import org.springframework.web.util.ContentCachingRequestWrapper;
 public class FluxHandlerMethodArgumentResolver
 		implements HandlerMethodArgumentResolver, Ordered {
 
+	private static Log logger = LogFactory
+			.getLog(FluxHandlerMethodArgumentResolver.class);
+
 	public static final String HANDLER = FluxHandlerMethodArgumentResolver.class.getName()
 			+ ".HANDLER";
 
@@ -76,16 +82,22 @@ public class FluxHandlerMethodArgumentResolver
 		List<Object> body;
 		ContentCachingRequestWrapper nativeRequest = new ContentCachingRequestWrapper(
 				webRequest.getNativeRequest(HttpServletRequest.class));
+		if (logger.isDebugEnabled()) {
+			logger.debug("Resolving request body into type: " + type);
+		}
 		if (isPlainText(webRequest) && CharSequence.class.isAssignableFrom(type)) {
 			body = Arrays.asList(StreamUtils.copyToString(nativeRequest.getInputStream(),
 					Charset.forName("UTF-8")));
 		}
 		else {
 			try {
-			body = mapper.readValue(nativeRequest.getInputStream(), mapper
-					.getTypeFactory().constructCollectionLikeType(ArrayList.class, type));
-			} catch (JsonMappingException e) {
-				body = Arrays.asList(mapper.readValue(nativeRequest.getContentAsByteArray(), type));
+				body = mapper.readValue(nativeRequest.getInputStream(),
+						mapper.getTypeFactory()
+								.constructCollectionLikeType(ArrayList.class, type));
+			}
+			catch (JsonMappingException e) {
+				body = Arrays.asList(
+						mapper.readValue(nativeRequest.getContentAsByteArray(), type));
 			}
 		}
 		return new FluxRequest<Object>(body);