Add FunctionFactoryMetadata interface for type discovery

The compiled functions implement that interface which means you can inspect
the signature of the method that created them and discover its
generic types.

README.adoc

@@ -156,7 +156,7 @@ curl -H "Accept: application/json" localhost:9001/words
 === Register a Consumer:
 
 ----
-./registerConsumer.sh -n print -f "System.out::println"
+./registerConsumer.sh -n print -f "f->f.subscribe(System.out::println)"
 ----
 
 === Run a REST Microservice using that Consumer:

scripts/registerConsumer.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-while getopts ":n:f:" opt; do
+while getopts ":n:f:t:" opt; do
   case $opt in
     n)
       NAME=$OPTARG
@@ -8,7 +8,10 @@ while getopts ":n:f:" opt; do
     f)
       FUNC=$OPTARG
       ;;
+    t)
+      TYPE=$OPTARG
+      ;;
     esac
 done
 
-curl -X POST -H "Content-Type: text/plain" -d $FUNC localhost:8080/consumer/$NAME
+curl -X POST -H "Content-Type: text/plain" -d $FUNC localhost:8080/consumer/$NAME?type=$TYPE

spring-cloud-function-compiler/pom.xml

@@ -44,6 +44,12 @@
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-web</artifactId>
+			<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
 		</dependency>
 	</dependencies>
 

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/CompilationResultFactory.java

@@ -23,7 +23,4 @@ public interface CompilationResultFactory<T> {
 
 	T getResult();
 
-	String getInputType();
-
-	String getOutputType();
 }

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/CompiledFunctionFactory.java

@@ -16,9 +16,12 @@
 
 package org.springframework.cloud.function.compiler;
 
+import java.lang.reflect.Method;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.springframework.cloud.function.compiler.java.CompilationResult;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Mark Fisher
@@ -33,18 +36,26 @@ public class CompiledFunctionFactory<T> implements CompilationResultFactory<T> {
 
 	private String outputType;
 
-	public CompiledFunctionFactory(String className, CompilationResult compilationResult) {
+	private Method method;
+
+	public CompiledFunctionFactory(String className,
+			CompilationResult compilationResult) {
 		List<Class<?>> clazzes = compilationResult.getCompiledClasses();
 		T result = null;
-		for (Class<?> clazz: clazzes) {
+		Method method = null;
+		for (Class<?> clazz : clazzes) {
 			if (clazz.getName().equals(className)) {
 				try {
 					@SuppressWarnings("unchecked")
-					CompilationResultFactory<T> factory = (CompilationResultFactory<T>) clazz.newInstance();
+					CompilationResultFactory<T> factory = (CompilationResultFactory<T>) clazz
+							.newInstance();
 					result = factory.getResult();
+					method = findFactoryMethod(clazz);
 				}
 				catch (Exception e) {
-					throw new IllegalArgumentException("Unexpected problem during retrieval of Function from compiled class", e);
+					throw new IllegalArgumentException(
+							"Unexpected problem during retrieval of Function from compiled class",
+							e);
 				}
 			}
 		}
@@ -52,13 +63,29 @@ public class CompiledFunctionFactory<T> implements CompilationResultFactory<T> {
 			throw new IllegalArgumentException("Failed to extract compilation result.");
 		}
 		this.result = result;
+		this.method = method;
 		this.generatedClassBytes = compilationResult.getClassBytes(className);
 	}
 
+	private Method findFactoryMethod(Class<?> clazz) {
+		AtomicReference<Method> method = new AtomicReference<>();
+		ReflectionUtils.doWithLocalMethods(clazz, m -> {
+			if (m.getName().equals("getResult")
+					&& m.getReturnType().getName().startsWith("java.util.function")) {
+				method.set(m);
+			}
+		});
+		return method.get();
+	}
+
 	public T getResult() {
 		return result;
 	}
 
+	public Method getFactoryMethod() {
+		return method;
+	}
+
 	public String getInputType() {
 		return inputType;
 	}

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/FunctionCompiler.java

@@ -45,6 +45,7 @@ public class FunctionCompiler<T, R> extends AbstractFunctionCompiler<Function<T,
 	protected CompiledFunctionFactory<Function<T, R>> postProcessCompiledFunctionFactory(CompiledFunctionFactory<Function<T, R>> factory) {
 		factory.setInputType(this.inputType);
 		factory.setOutputType(this.outputType);
+		
 		return factory;
 	}
 }

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/app/CompiledFunctionRegistry.java

@@ -94,8 +94,8 @@ public class CompiledFunctionRegistry {
 		}
 	}
 
-	public void registerConsumer(String name, String consumer) {
-		CompiledFunctionFactory<?> factory = this.consumerCompiler.compile(name, consumer);
+	public void registerConsumer(String name, String consumer, String type) {
+		CompiledFunctionFactory<?> factory = this.consumerCompiler.compile(name, consumer, type);
 		File file = new File(this.consumerDirectory, fileName(name));
 		try {
 			FileCopyUtils.copy(factory.getGeneratedClassBytes(), file);

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/app/CompilerController.java

@@ -44,7 +44,8 @@ public class CompilerController {
 	}
 
 	@PostMapping(path = "/consumer/{name}")
-	public void registerConsumer(@PathVariable String name, @RequestBody String lambda) {
-		this.registry.registerConsumer(name, lambda);
+	public void registerConsumer(@PathVariable String name, @RequestBody String lambda,
+			@RequestParam(defaultValue="Flux<String>") String type) {
+		this.registry.registerConsumer(name, lambda, type);
 	}
 }

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/config/FunctionProxyApplicationListener.java

@@ -115,6 +115,9 @@ public class FunctionProxyApplicationListener implements ApplicationListener<App
 		else if ("consumer".equals(type.toLowerCase())) {
 			proxyClass = LambdaCompilingConsumer.class;
 			args.addGenericArgumentValue(this.consumerCompiler);
+			if (inputType != null) {
+				props.add("typeParameterizations", inputType);
+			}
 		}
 		else {
 			proxyClass = LambdaCompilingFunction.class;

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/proxy/AbstractByteCodeLoadingProxy.java

@@ -16,12 +16,16 @@
 
 package org.springframework.cloud.function.compiler.proxy;
 
+import java.lang.reflect.Method;
+import java.util.concurrent.atomic.AtomicReference;
+
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.cloud.function.compiler.CompilationResultFactory;
 import org.springframework.cloud.function.compiler.FunctionCompiler;
 import org.springframework.cloud.function.compiler.java.SimpleClassLoader;
 import org.springframework.core.io.Resource;
 import org.springframework.util.FileCopyUtils;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * @author Mark Fisher
@@ -36,6 +40,8 @@ public abstract class AbstractByteCodeLoadingProxy<T> implements InitializingBea
 
 	private final SimpleClassLoader classLoader = new SimpleClassLoader(AbstractByteCodeLoadingProxy.class.getClassLoader());
 
+	private Method method;
+
 	public AbstractByteCodeLoadingProxy(Resource resource, Class<?> type) {
 		this.resource = resource;
 		this.type = type;
@@ -45,28 +51,37 @@ public abstract class AbstractByteCodeLoadingProxy<T> implements InitializingBea
 	@SuppressWarnings("unchecked")
 	public void afterPropertiesSet() throws Exception {
 		byte[] bytes = FileCopyUtils.copyToByteArray(this.resource.getInputStream());
-		String functionName = this.resource.getFilename().replaceAll(".fun$", "");
+		String filename = this.resource.getFilename();
+		String functionName = filename == null ? type.getSimpleName() : filename.replaceAll(".fun$", "");
 		String firstLetter = functionName.substring(0, 1).toUpperCase();
 		String upperCasedName = (functionName.length() > 1) ? firstLetter + functionName.substring(1) : firstLetter;
 		String className = String.format("%s.%s%sFactory", FunctionCompiler.class.getPackage().getName(), upperCasedName, this.type.getSimpleName()); 
 		Class<?> factoryClass = this.classLoader.defineClass(className, bytes);
 		try {
 			this.factory = (CompilationResultFactory<T>) factoryClass.newInstance();
+			this.method = findFactoryMethod(factoryClass);
 		}
 		catch (InstantiationException | IllegalAccessException e) {
 			throw new IllegalArgumentException("failed to load Function byte code", e);
 		}
 	}
 
+	private Method findFactoryMethod(Class<?> clazz) {
+		AtomicReference<Method> method = new AtomicReference<>();
+		ReflectionUtils.doWithLocalMethods(clazz, m -> {
+			if (m.getName().equals("getResult")
+					&& m.getReturnType().getName().startsWith("java.util.function")) {
+				method.set(m);
+			}
+		});
+		return method.get();
+	}
+
 	public final T getTarget() {
 		return this.factory.getResult();
 	}
 
-	public String getInputType() {
-		return this.factory.getInputType();
-	}
-
-	public String getOutputType() {
-		return this.factory.getOutputType();
+	public Method getFactoryMethod() {
+		return this.method;
 	}
 }

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/proxy/AbstractLambdaCompilingProxy.java

@@ -17,11 +17,13 @@
 package org.springframework.cloud.function.compiler.proxy;
 
 import java.io.InputStreamReader;
+import java.lang.reflect.Method;
 
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.cloud.function.compiler.AbstractFunctionCompiler;
 import org.springframework.cloud.function.compiler.CompiledFunctionFactory;
+import org.springframework.cloud.function.support.FunctionFactoryMetadata;
 import org.springframework.core.io.Resource;
 import org.springframework.util.Assert;
 import org.springframework.util.FileCopyUtils;
@@ -29,7 +31,7 @@ import org.springframework.util.FileCopyUtils;
 /**
  * @author Mark Fisher
  */
-public class AbstractLambdaCompilingProxy<T> implements InitializingBean, BeanNameAware {
+public class AbstractLambdaCompilingProxy<T> implements InitializingBean, BeanNameAware, FunctionFactoryMetadata {
 
 	private final Resource resource;
 
@@ -67,11 +69,7 @@ public class AbstractLambdaCompilingProxy<T> implements InitializingBean, BeanNa
 		return this.factory.getResult();
 	}
 
-	public final String getInputType() {
-		return this.factory.getInputType();
-	}
-
-	public final String getOutputType() {
-		return this.factory.getOutputType();
+	public Method getFactoryMethod() {
+		return this.factory.getFactoryMethod();
 	}
 }

spring-cloud-function-compiler/test/java/org/springframework/cloud/function/compiler/ConsumerCompilerTests.java

@@ -0,0 +1,48 @@
+/*
+ * 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.compiler;
+
+import java.util.function.Consumer;
+
+import org.junit.Test;
+
+import org.springframework.cloud.function.support.FunctionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class ConsumerCompilerTests {
+
+	@Test
+	public void consumesFluxString() {
+		CompiledFunctionFactory<Consumer<String>> compiled = new ConsumerCompiler<String>(
+				String.class.getName()).compile("foos",
+						"flux -> flux.subscribe(System.out::println)", "Flux<String>");
+		assertThat(FunctionUtils.isFluxConsumer(compiled.getFactoryMethod())).isTrue();
+	}
+
+	@Test
+	public void consumesString() {
+		CompiledFunctionFactory<Consumer<String>> compiled = new ConsumerCompiler<String>(
+				String.class.getName()).compile("foos", "System.out::println", "String");
+		assertThat(FunctionUtils.isFluxConsumer(compiled.getFactoryMethod())).isFalse();
+	}
+
+}

spring-cloud-function-compiler/test/java/org/springframework/cloud/function/compiler/FunctionCompilerTests.java

@@ -0,0 +1,49 @@
+/*
+ * 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.compiler;
+
+import java.util.function.Function;
+
+import org.junit.Test;
+
+import org.springframework.cloud.function.support.FunctionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class FunctionCompilerTests {
+
+	@Test
+	public void transformsFluxString() {
+		CompiledFunctionFactory<Function<String, String>> compiled = new FunctionCompiler<String, String>(
+				String.class.getName()).compile("foos",
+						"flux -> flux.map(v -> v.toUpperCase())", "Flux<String>", "Flux<String>");
+		assertThat(FunctionUtils.isFluxFunction(compiled.getFactoryMethod())).isTrue();
+	}
+
+	@Test
+	public void transformsString() {
+		CompiledFunctionFactory<Function<String, String>> compiled = new FunctionCompiler<String, String>(
+				String.class.getName()).compile("foos", "v -> v.toUpperCase()", "String", "String");
+		assertThat(FunctionUtils.isFluxFunction(compiled.getFactoryMethod())).isFalse();
+		assertThat(compiled.getResult().apply("hello")).isEqualTo("HELLO");
+	}
+
+}

spring-cloud-function-compiler/test/java/org/springframework/cloud/function/compiler/SupplierCompilerTests.java

@@ -0,0 +1,50 @@
+/*
+ * 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.compiler;
+
+import java.util.function.Supplier;
+
+import org.junit.Test;
+
+import org.springframework.cloud.function.support.FunctionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class SupplierCompilerTests {
+
+	@Test
+	public void supppliesFluxString() {
+		CompiledFunctionFactory<Supplier<String>> compiled = new SupplierCompiler<String>(
+				String.class.getName()).compile("foos",
+						"() -> Flux.just(\"foo\", \"bar\")", "Flux<String>");
+		assertThat(FunctionUtils.isFluxSupplier(compiled.getFactoryMethod())).isTrue();
+	}
+
+	@Test
+	public void supppliesString() {
+		CompiledFunctionFactory<Supplier<String>> compiled = new SupplierCompiler<String>(
+				String.class.getName()).compile("foos",
+						"() -> \"foo\"", "String");
+		assertThat(FunctionUtils.isFluxSupplier(compiled.getFactoryMethod())).isFalse();
+		assertThat(compiled.getResult().get()).isEqualTo("foo");
+	}
+
+}

spring-cloud-function-compiler/test/java/org/springframework/cloud/function/compiler/proxy/ByteCodeLoadingFunctionTests.java

@@ -0,0 +1,106 @@
+/*
+ * 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.compiler.proxy;
+
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.junit.Test;
+
+import org.springframework.cloud.function.compiler.CompiledFunctionFactory;
+import org.springframework.cloud.function.compiler.ConsumerCompiler;
+import org.springframework.cloud.function.compiler.FunctionCompiler;
+import org.springframework.cloud.function.compiler.SupplierCompiler;
+import org.springframework.cloud.function.support.FunctionUtils;
+import org.springframework.core.io.ByteArrayResource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class ByteCodeLoadingFunctionTests {
+
+	@Test
+	public void compileConsumer() throws Exception {
+		CompiledFunctionFactory<Consumer<String>> compiled = new ConsumerCompiler<String>(
+				String.class.getName()).compile("foos", "System.out::println", "String");
+		ByteArrayResource resource = new ByteArrayResource(compiled.getGeneratedClassBytes(), "foos") {
+			@Override
+			public String getFilename() {
+				return "foos.fun";
+			}
+		};
+		ByteCodeLoadingConsumer<String> consumer = new ByteCodeLoadingConsumer<>(resource);
+		consumer.afterPropertiesSet();
+		assertThat(FunctionUtils.isFluxConsumer(consumer.getFactoryMethod())).isFalse();
+		consumer.accept("foo");
+	}
+
+	@Test
+	public void compileSupplier() throws Exception {
+		CompiledFunctionFactory<Supplier<String>> compiled = new SupplierCompiler<String>(
+				String.class.getName()).compile("foos", "() -> \"foo\"", "String");
+		ByteArrayResource resource = new ByteArrayResource(compiled.getGeneratedClassBytes(), "foos") {
+			@Override
+			public String getFilename() {
+				return "foos.fun";
+			}
+		};
+		ByteCodeLoadingSupplier<String> supplier = new ByteCodeLoadingSupplier<>(resource);
+		supplier.afterPropertiesSet();
+		assertThat(FunctionUtils.isFluxSupplier(supplier.getFactoryMethod())).isFalse();
+		assertThat(supplier.get()).isEqualTo("foo");
+	}
+
+	@Test
+	public void compileFunction() throws Exception {
+		CompiledFunctionFactory<Function<String, String>> compiled = new FunctionCompiler<String, String>(
+				String.class.getName()).compile("foos", "v -> v.toUpperCase()", "String", "String");
+		ByteArrayResource resource = new ByteArrayResource(compiled.getGeneratedClassBytes(), "foos") {
+			@Override
+			public String getFilename() {
+				return "foos.fun";
+			}
+		};
+		ByteCodeLoadingFunction<String, String> function = new ByteCodeLoadingFunction<>(resource);
+		function.afterPropertiesSet();
+		assertThat(FunctionUtils.isFluxFunction(function.getFactoryMethod())).isFalse();
+		assertThat(function.apply("foo")).isEqualTo("FOO");
+	}
+
+	@Test
+	public void compileFluxFunction() throws Exception {
+		CompiledFunctionFactory<Function<Flux<String>, Flux<String>>> compiled = new FunctionCompiler<Flux<String>, Flux<String>>(
+				String.class.getName()).compile("foos", "flux -> flux.map(v -> v.toUpperCase())", "Flux<String>", "Flux<String>");
+		ByteArrayResource resource = new ByteArrayResource(compiled.getGeneratedClassBytes(), "foos") {
+			@Override
+			public String getFilename() {
+				return "foos.fun";
+			}
+		};
+		ByteCodeLoadingFunction<Flux<String>, Flux<String>> function = new ByteCodeLoadingFunction<>(resource);
+		function.afterPropertiesSet();
+		assertThat(FunctionUtils.isFluxFunction(function.getFactoryMethod())).isTrue();
+		assertThat(function.apply(Flux.just("foo")).blockFirst()).isEqualTo("FOO");
+	}
+
+}

spring-cloud-function-context/pom.xml

@@ -42,5 +42,11 @@
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-function-compiler</artifactId>
+			<version>${spring-cloud-function.version}</version>
+			<scope>test</scope>
+		</dependency>
 	</dependencies>
 </project>

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

@@ -35,7 +35,7 @@ import java.util.function.Function;
 import java.util.function.Supplier;
 
 import org.springframework.beans.BeansException;
-import org.springframework.beans.factory.ListableBeanFactory;
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.beans.factory.config.BeanDefinition;
@@ -51,6 +51,7 @@ import org.springframework.cloud.function.registry.FunctionCatalog;
 import org.springframework.cloud.function.support.FluxConsumer;
 import org.springframework.cloud.function.support.FluxFunction;
 import org.springframework.cloud.function.support.FluxSupplier;
+import org.springframework.cloud.function.support.FunctionFactoryMetadata;
 import org.springframework.cloud.function.support.FunctionUtils;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -391,7 +392,7 @@ public class ContextFunctionCatalogAutoConfiguration {
 		private Class<?> findType(String name, AbstractBeanDefinition definition,
 				ParamType paramType) {
 			Object source = definition.getSource();
-			Type param;
+			Type param = null;
 			// Start by assuming output -> Function
 			int index = paramType.isOutput() ? 1 : 0;
 			if (source instanceof StandardMethodMetadata) {
@@ -422,18 +423,26 @@ public class ContextFunctionCatalogAutoConfiguration {
 				if (resolvable != null) {
 					param = resolvable.getGeneric(index).getGeneric(0).getType();
 				}
-				else {
-					// TODO: compiled functions (only work as String -> String)
-					if (paramType.isWrapper() && !Consumer.class.isAssignableFrom(definition.getBeanClass())) {
-						return Flux.class;
+				else if (registry instanceof BeanFactory) {
+					Object bean = ((BeanFactory) registry).getBean(name);
+					if (bean instanceof FunctionFactoryMetadata) {
+						FunctionFactoryMetadata factory = (FunctionFactoryMetadata) bean;
+						Type type = factory.getFactoryMethod().getGenericReturnType();
+						param = extractType(type, paramType, index);
 					}
-					return String.class;
 				}
 			}
 			if (param instanceof ParameterizedType) {
 				ParameterizedType concrete = (ParameterizedType) param;
 				param = concrete.getRawType();
 			}
+			if (param == null) {
+				// Last ditch attempt to guess: Flux<String>
+				if (paramType.isWrapper()) {
+					return Flux.class;
+				}
+				return String.class;
+			}
 			return (Class<?>) param;
 		}
 

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

@@ -30,12 +30,16 @@ 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.compiler.CompiledFunctionFactory;
+import org.springframework.cloud.function.compiler.FunctionCompiler;
 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 org.springframework.core.io.FileSystemResource;
+import org.springframework.util.StreamUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -143,12 +147,35 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 		assertThat(catalog.lookupFunction("other")).isInstanceOf(Function.class);
 	}
 
-	private void create(Class<?>... types) {
-		context = new SpringApplicationBuilder((Object[]) types).run();
+	@Test
+	public void compiledFunction() throws Exception {
+		CompiledFunctionFactory<Function<String, String>> compiled = new FunctionCompiler<String, String>(
+				String.class.getName()).compile("foos", "v -> v.toUpperCase()", "String",
+						"String");
+		FileSystemResource resource = new FileSystemResource("target/foos.fun");
+		StreamUtils.copy(compiled.getGeneratedClassBytes(), resource.getOutputStream());
+		create(EmptyConfiguration.class,
+				"spring.cloud.function.import.foos.location=file:./target/foos.fun");
+		assertThat(context.getBean("foos")).isInstanceOf(Function.class);
+		assertThat(catalog.lookupFunction("foos")).isInstanceOf(Function.class);
+		assertThat(inspector.getInputWrapper("foos")).isEqualTo(String.class);
+	}
+
+	private void create(Class<?> type, String... props) {
+		create(new Class<?>[] { type }, props);
+	}
+
+	private void create(Class<?>[] types, String... props) {
+		context = new SpringApplicationBuilder((Object[]) types).properties(props).run();
 		catalog = context.getBean(InMemoryFunctionCatalog.class);
 		inspector = context.getBean(FunctionInspector.class);
 	}
 
+	@EnableAutoConfiguration
+	@Configuration
+	protected static class EmptyConfiguration {
+	}
+
 	@EnableAutoConfiguration
 	@Configuration
 	protected static class SimpleConfiguration {
@@ -188,7 +215,7 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 
 	@EnableAutoConfiguration
 	@Configuration
-	@ComponentScan(basePackageClasses=GenericFunction.class)
+	@ComponentScan(basePackageClasses = GenericFunction.class)
 	protected static class ComponentScanConfiguration {
 	}
 
@@ -237,4 +264,3 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 	}
 
 }
-

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

@@ -23,13 +23,12 @@ import java.util.function.Consumer;
  *
  * @param <T> output type of target Consumer
  */
-public interface ConsumerProxy<T> extends Consumer<T> {
+public interface ConsumerProxy<T> extends Consumer<T>, FunctionFactoryMetadata {
 
 	default boolean isFluxConsumer() {
-		return FunctionUtils.isFluxConsumer(getTarget());
+		return FunctionUtils.isFluxConsumer(getFactoryMethod());
 	}
 
 	Consumer<T> getTarget();
 
-	String getInputType();
 }

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

@@ -0,0 +1,30 @@
+/*
+ * 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.support;
+
+import java.lang.reflect.Method;
+
+/**
+ * @author Dave Syer
+ *
+ * @param <T>
+ */
+public interface FunctionFactoryMetadata {
+
+	Method getFactoryMethod();
+
+}

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

@@ -24,15 +24,12 @@ import java.util.function.Function;
  * @param <T> input type of target Function
  * @param <R> output type of target Function
  */
-public interface FunctionProxy<T, R> extends Function<T, R> {
+public interface FunctionProxy<T, R> extends Function<T, R>, FunctionFactoryMetadata {
 
 	default boolean isFluxFunction() {
-		return FunctionUtils.isFluxFunction(getTarget());
+		return FunctionUtils.isFluxFunction(getFactoryMethod());
 	}
 
 	Function<T, R> getTarget();
 
-	String getInputType();
-
-	String getOutputType();
 }

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

@@ -118,4 +118,51 @@ public abstract class FunctionUtils {
 		}
 		return typeNames.toArray(new String[typeNames.size()]);
 	}
+
+	public static boolean isFluxSupplier(Method method) {
+		String[] types = getParameterizedTypeNamesForMethod(method, Supplier.class);
+		if (ObjectUtils.isEmpty(types)) {
+			return false;
+		}
+		return (types[0].startsWith(FLUX_CLASS_NAME));
+	}
+
+	public static boolean isFluxConsumer(Method method) {
+		String[] types = getParameterizedTypeNamesForMethod(method, Consumer.class);
+		if (ObjectUtils.isEmpty(types)) {
+			return false;
+		}
+		return (types[0].startsWith(FLUX_CLASS_NAME));
+	}
+
+	public static boolean isFluxFunction(Method method) {
+		String[] types = getParameterizedTypeNamesForMethod(method, Function.class);
+		if (ObjectUtils.isEmpty(types)) {
+			return false;
+		}
+		if (ObjectUtils.isEmpty(types) || types.length != 2) {
+			return false;
+		}
+		return (types[0].startsWith(FLUX_CLASS_NAME)
+				&& types[1].startsWith(FLUX_CLASS_NAME));
+	}
+
+	private static String[] getParameterizedTypeNamesForMethod(Method method,
+			Class<?> interfaceClass) {
+		Type genericInterface = method.getGenericReturnType();
+		if ((genericInterface instanceof ParameterizedType) && interfaceClass
+				.getTypeName().equals(((ParameterizedType) genericInterface)
+						.getRawType().getTypeName())) {
+			ParameterizedType type = (ParameterizedType) genericInterface;
+			Type[] args = type.getActualTypeArguments();
+			if (args != null) {
+				String[] typeNames = new String[args.length];
+				for (int i = 0; i < args.length; i++) {
+					typeNames[i] = args[i].getTypeName();
+				}
+				return typeNames;
+			}
+		}
+		return new String[0];
+	}
 }

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

@@ -23,13 +23,11 @@ import java.util.function.Supplier;
  *
  * @param <T> output type of target Supplier
  */
-public interface SupplierProxy<T> extends Supplier<T> {
+public interface SupplierProxy<T> extends Supplier<T>, FunctionFactoryMetadata {
 
 	default boolean isFluxSupplier() {
-		return FunctionUtils.isFluxSupplier(getTarget());
+		return FunctionUtils.isFluxSupplier(getFactoryMethod());
 	}
 
 	Supplier<T> getTarget();
-
-	String getOutputType();
 }

spring-cloud-function-core/src/test/java/org/springframework/cloud/function/gateway/FunctionUtilsTests.java

@@ -0,0 +1,77 @@
+/*
+ * 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.gateway;
+
+import java.lang.reflect.Method;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.junit.Test;
+
+import org.springframework.cloud.function.support.FunctionUtils;
+import org.springframework.util.ReflectionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class FunctionUtilsTests {
+
+	@Test
+	public void isFluxConsumer() {
+		Method method = ReflectionUtils.findMethod(FunctionUtilsTests.class, "fluxConsumer");
+		assertThat(FunctionUtils.isFluxConsumer(method)).isTrue();
+		assertThat(FunctionUtils.isFluxSupplier(method)).isFalse();
+		assertThat(FunctionUtils.isFluxFunction(method)).isFalse();
+	}
+	
+	@Test
+	public void isFluxSupplier() {
+		Method method = ReflectionUtils.findMethod(FunctionUtilsTests.class, "fluxSupplier");
+		assertThat(FunctionUtils.isFluxSupplier(method)).isTrue();
+		assertThat(FunctionUtils.isFluxConsumer(method)).isFalse();
+		assertThat(FunctionUtils.isFluxFunction(method)).isFalse();
+	}
+
+	@Test
+	public void isFluxFunction() {
+		Method method = ReflectionUtils.findMethod(FunctionUtilsTests.class, "fluxFunction");
+		assertThat(FunctionUtils.isFluxFunction(method)).isTrue();
+		assertThat(FunctionUtils.isFluxSupplier(method)).isFalse();
+		assertThat(FunctionUtils.isFluxConsumer(method)).isFalse();
+	}
+	
+	public Function<Flux<Foo>, Flux<Foo>> fluxFunction() {
+		return foos -> foos.map(foo -> new Foo());
+	}
+
+	public Supplier<Flux<Foo>> fluxSupplier() {
+		return () -> Flux.just(new Foo());
+	}
+
+	public Consumer<Flux<Foo>> fluxConsumer() {
+		return flux -> flux.subscribe(System.out::println);
+	}
+	
+	class Foo {}
+
+}

spring-cloud-function-samples/spring-cloud-function-sample-compiler/src/test/java/com/example/SampleCompiledConsumerTests.java

@@ -36,6 +36,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 @SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT,
 		properties = {
 				"spring.cloud.function.compile.test.lambda=com.example.SampleCompiledConsumerTests.Reference::set",
+				"spring.cloud.function.compile.test.inputType=String",
 				"spring.cloud.function.compile.test.type=consumer"
 		})
 public class SampleCompiledConsumerTests {

spring-cloud-function-samples/spring-cloud-function-sample-compiler/src/test/java/com/example/SampleCompiledFunctionTests.java

@@ -31,8 +31,10 @@ import static org.assertj.core.api.Assertions.assertThat;
  * @author Mark Fisher
  */
 @RunWith(SpringRunner.class)
-@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT,
-		properties = "spring.cloud.function.compile.test.lambda=f->f.map(s->s+\"!!!\")")
+@SpringBootTest(webEnvironment = WebEnvironment.RANDOM_PORT, properties = {
+		"spring.cloud.function.compile.test.lambda=f->f.map(s->s+\"!!!\")",
+		"spring.cloud.function.compile.test.inputType=Flux<String>",
+		"spring.cloud.function.compile.test.outputType=Flux<String>" })
 public class SampleCompiledFunctionTests {
 
 	@LocalServerPort
@@ -41,8 +43,8 @@ public class SampleCompiledFunctionTests {
 	@Test
 	public void lowercase() {
 		assertThat(new TestRestTemplate().postForObject(
-				"http://localhost:" + port + "/test", "it works",
-				String.class)).isEqualTo("it works!!!");
+				"http://localhost:" + port + "/test", "it works", String.class))
+						.isEqualTo("it works!!!");
 	}
 
 }