FunctionUtils polishing

- attempted to make code more functional (eat our own dog food) and concise
- addressed PR comment
- additional cleanup/polishing of FunctionUtils and related classes
- Removed Function/Supplier/ConsumerProxy classes by extending type info on their super interface
- Renamed FunctionUtils to FunctionFactoryUtils
- Added javadoc to FunctionFactoryUtils to explain its design considerations as well as what it can and can not doi

Fixes gh-90

spring-cloud-function-adapters/spring-cloud-function-adapter-openwhisk/src/main/java/org/springframework/cloud/function/adapter/openwhisk/FunctionInitializer.java

@@ -24,8 +24,8 @@ import java.util.function.Supplier;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.function.context.FunctionInspector;
 import org.springframework.cloud.function.registry.FunctionCatalog;
+import org.springframework.cloud.function.support.FunctionFactoryUtils;
 import org.springframework.cloud.function.support.FluxFunction;
-import org.springframework.cloud.function.support.FunctionUtils;
 
 import reactor.core.publisher.Flux;
 
@@ -61,7 +61,7 @@ public class FunctionInitializer {
 		String type = this.properties.getType();
 		if ("function".equals(type)) {
 			this.function = this.catalog.lookupFunction(name);
-			if (this.function != null && !FunctionUtils.isFluxFunction(this.function)) {
+			if (this.function != null && !FunctionFactoryUtils.isFluxFunction(this.function)) {
 				// TODO: this shouldn't be necessary
 				this.function = new FluxFunction(this.function);
 			}

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

@@ -31,13 +31,13 @@ import org.springframework.util.ReflectionUtils;
 /**
  * @author Mark Fisher
  */
-public abstract class AbstractByteCodeLoadingProxy<T> implements InitializingBean, FunctionFactoryMetadata {
+public abstract class AbstractByteCodeLoadingProxy<T> implements InitializingBean, FunctionFactoryMetadata<T> {
 
 	private final Resource resource;
 
 	private final Class<?> type;
 
-	private CompilationResultFactory<T> factory; 
+	private CompilationResultFactory<T> factory;
 
 	private final SimpleClassLoader classLoader = new SimpleClassLoader(AbstractByteCodeLoadingProxy.class.getClassLoader());
 
@@ -56,7 +56,7 @@ public abstract class AbstractByteCodeLoadingProxy<T> implements InitializingBea
 		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()); 
+		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();
@@ -78,6 +78,7 @@ public abstract class AbstractByteCodeLoadingProxy<T> implements InitializingBea
 		return method.get();
 	}
 
+	@Override
 	public final T getTarget() {
 		return this.factory.getResult();
 	}

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

@@ -31,7 +31,7 @@ import org.springframework.util.FileCopyUtils;
 /**
  * @author Mark Fisher
  */
-public class AbstractLambdaCompilingProxy<T> implements InitializingBean, BeanNameAware, FunctionFactoryMetadata {
+public class AbstractLambdaCompilingProxy<T> implements InitializingBean, BeanNameAware, FunctionFactoryMetadata<T> {
 
 	private final Resource resource;
 
@@ -65,6 +65,7 @@ public class AbstractLambdaCompilingProxy<T> implements InitializingBean, BeanNa
 		this.factory = this.compiler.compile(this.beanName, lambda, this.typeParameterizations);
 	}
 
+	@Override
 	public final T getTarget() {
 		return this.factory.getResult();
 	}

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

@@ -18,7 +18,7 @@ package org.springframework.cloud.function.compiler.proxy;
 
 import java.util.function.Consumer;
 
-import org.springframework.cloud.function.support.ConsumerProxy;
+import org.springframework.cloud.function.support.FunctionFactoryMetadata;
 import org.springframework.core.io.Resource;
 
 /**
@@ -26,7 +26,7 @@ import org.springframework.core.io.Resource;
  *
  * @param <T> type
  */
-public class ByteCodeLoadingConsumer<T> extends AbstractByteCodeLoadingProxy<Consumer<T>> implements ConsumerProxy<T> {
+public class ByteCodeLoadingConsumer<T> extends AbstractByteCodeLoadingProxy<Consumer<T>> implements FunctionFactoryMetadata<Consumer<T>>, Consumer<T> {
 
 	public ByteCodeLoadingConsumer(Resource resource) {
 		super(resource, Consumer.class);

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

@@ -18,7 +18,7 @@ package org.springframework.cloud.function.compiler.proxy;
 
 import java.util.function.Function;
 
-import org.springframework.cloud.function.support.FunctionProxy;
+import org.springframework.cloud.function.support.FunctionFactoryMetadata;
 import org.springframework.core.io.Resource;
 
 /**
@@ -27,7 +27,7 @@ import org.springframework.core.io.Resource;
  * @param <T> Function input type
  * @param <R> Function result type
  */
-public class ByteCodeLoadingFunction<T, R> extends AbstractByteCodeLoadingProxy<Function<T, R>> implements FunctionProxy<T, R> {
+public class ByteCodeLoadingFunction<T, R> extends AbstractByteCodeLoadingProxy<Function<T, R>> implements FunctionFactoryMetadata<Function<T, R>>, Function<T, R> {
 
 	public ByteCodeLoadingFunction(Resource resource) {
 		super(resource, Function.class);

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

@@ -18,7 +18,7 @@ package org.springframework.cloud.function.compiler.proxy;
 
 import java.util.function.Supplier;
 
-import org.springframework.cloud.function.support.SupplierProxy;
+import org.springframework.cloud.function.support.FunctionFactoryMetadata;
 import org.springframework.core.io.Resource;
 
 /**
@@ -26,7 +26,7 @@ import org.springframework.core.io.Resource;
  *
  * @param <T> type
  */
-public class ByteCodeLoadingSupplier<T> extends AbstractByteCodeLoadingProxy<Supplier<T>> implements SupplierProxy<T> {
+public class ByteCodeLoadingSupplier<T> extends AbstractByteCodeLoadingProxy<Supplier<T>> implements FunctionFactoryMetadata<Supplier<T>>, Supplier<T> {
 
 	public ByteCodeLoadingSupplier(Resource resource) {
 		super(resource, Supplier.class);

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

@@ -19,13 +19,13 @@ package org.springframework.cloud.function.compiler.proxy;
 import java.util.function.Function;
 
 import org.springframework.cloud.function.compiler.FunctionCompiler;
-import org.springframework.cloud.function.support.FunctionProxy;
+import org.springframework.cloud.function.support.FunctionFactoryMetadata;
 import org.springframework.core.io.Resource;
 
 /**
  * @author Mark Fisher
  */
-public class LambdaCompilingFunction<T, R> extends AbstractLambdaCompilingProxy<Function<T, R>> implements FunctionProxy<T, R> {
+public class LambdaCompilingFunction<T, R> extends AbstractLambdaCompilingProxy<Function<T, R>> implements FunctionFactoryMetadata<Function<T, R>>, Function<T, R> {
 
 	public LambdaCompilingFunction(Resource resource, FunctionCompiler<T, R> compiler) {
 		super(resource, compiler);

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

@@ -19,13 +19,13 @@ package org.springframework.cloud.function.compiler.proxy;
 import java.util.function.Supplier;
 
 import org.springframework.cloud.function.compiler.SupplierCompiler;
-import org.springframework.cloud.function.support.SupplierProxy;
+import org.springframework.cloud.function.support.FunctionFactoryMetadata;
 import org.springframework.core.io.Resource;
 
 /**
  * @author Mark Fisher
  */
-public class LambdaCompilingSupplier<T> extends AbstractLambdaCompilingProxy<Supplier<T>> implements SupplierProxy<T> {
+public class LambdaCompilingSupplier<T> extends AbstractLambdaCompilingProxy<Supplier<T>> implements FunctionFactoryMetadata<Supplier<T>>, Supplier<T> {
 
 	public LambdaCompilingSupplier(Resource resource, SupplierCompiler<T> compiler) {
 		super(resource, compiler);

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

@@ -20,7 +20,7 @@ import java.util.function.Consumer;
 
 import org.junit.Test;
 
-import org.springframework.cloud.function.support.FunctionUtils;
+import org.springframework.cloud.function.support.FunctionFactoryUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -35,14 +35,14 @@ public class ConsumerCompilerTests {
 		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();
+		assertThat(FunctionFactoryUtils.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();
+		assertThat(FunctionFactoryUtils.isFluxConsumer(compiled.getFactoryMethod())).isFalse();
 	}
 
 }

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

@@ -20,7 +20,7 @@ import java.util.function.Function;
 
 import org.junit.Test;
 
-import org.springframework.cloud.function.support.FunctionUtils;
+import org.springframework.cloud.function.support.FunctionFactoryUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -35,14 +35,14 @@ public class FunctionCompilerTests {
 		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();
+		assertThat(FunctionFactoryUtils.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(FunctionFactoryUtils.isFluxFunction(compiled.getFactoryMethod())).isFalse();
 		assertThat(compiled.getResult().apply("hello")).isEqualTo("HELLO");
 	}
 

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

@@ -20,7 +20,7 @@ import java.util.function.Supplier;
 
 import org.junit.Test;
 
-import org.springframework.cloud.function.support.FunctionUtils;
+import org.springframework.cloud.function.support.FunctionFactoryUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -35,7 +35,7 @@ public class SupplierCompilerTests {
 		CompiledFunctionFactory<Supplier<String>> compiled = new SupplierCompiler<String>(
 				String.class.getName()).compile("foos",
 						"() -> Flux.just(\"foo\", \"bar\")", "Flux<String>");
-		assertThat(FunctionUtils.isFluxSupplier(compiled.getFactoryMethod())).isTrue();
+		assertThat(FunctionFactoryUtils.isFluxSupplier(compiled.getFactoryMethod())).isTrue();
 	}
 
 	@Test
@@ -43,7 +43,7 @@ public class SupplierCompilerTests {
 		CompiledFunctionFactory<Supplier<String>> compiled = new SupplierCompiler<String>(
 				String.class.getName()).compile("foos",
 						"() -> \"foo\"", "String");
-		assertThat(FunctionUtils.isFluxSupplier(compiled.getFactoryMethod())).isFalse();
+		assertThat(FunctionFactoryUtils.isFluxSupplier(compiled.getFactoryMethod())).isFalse();
 		assertThat(compiled.getResult().get()).isEqualTo("foo");
 	}
 

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

@@ -27,7 +27,7 @@ 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.FunctionFactoryMetadata;
-import org.springframework.cloud.function.support.FunctionUtils;
+import org.springframework.cloud.function.support.FunctionFactoryUtils;
 import org.springframework.core.io.ByteArrayResource;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -53,7 +53,7 @@ public class ByteCodeLoadingFunctionTests {
 		ByteCodeLoadingConsumer<String> consumer = new ByteCodeLoadingConsumer<>(resource);
 		consumer.afterPropertiesSet();
 		assertThat(consumer instanceof FunctionFactoryMetadata);
-		assertThat(FunctionUtils.isFluxConsumer(consumer.getFactoryMethod())).isFalse();
+		assertThat(FunctionFactoryUtils.isFluxConsumer(consumer.getFactoryMethod())).isFalse();
 		consumer.accept("foo");
 	}
 
@@ -70,7 +70,7 @@ public class ByteCodeLoadingFunctionTests {
 		ByteCodeLoadingSupplier<String> supplier = new ByteCodeLoadingSupplier<>(resource);
 		supplier.afterPropertiesSet();
 		assertThat(supplier instanceof FunctionFactoryMetadata);
-		assertThat(FunctionUtils.isFluxSupplier(supplier.getFactoryMethod())).isFalse();
+		assertThat(FunctionFactoryUtils.isFluxSupplier(supplier.getFactoryMethod())).isFalse();
 		assertThat(supplier.get()).isEqualTo("foo");
 	}
 
@@ -87,7 +87,7 @@ public class ByteCodeLoadingFunctionTests {
 		ByteCodeLoadingFunction<String, String> function = new ByteCodeLoadingFunction<>(resource);
 		function.afterPropertiesSet();
 		assertThat(function instanceof FunctionFactoryMetadata);
-		assertThat(FunctionUtils.isFluxFunction(function.getFactoryMethod())).isFalse();
+		assertThat(FunctionFactoryUtils.isFluxFunction(function.getFactoryMethod())).isFalse();
 		assertThat(function.apply("foo")).isEqualTo("FOO");
 	}
 
@@ -104,7 +104,7 @@ public class ByteCodeLoadingFunctionTests {
 		ByteCodeLoadingFunction<Flux<String>, Flux<String>> function = new ByteCodeLoadingFunction<>(resource);
 		function.afterPropertiesSet();
 		assertThat(function instanceof FunctionFactoryMetadata);
-		assertThat(FunctionUtils.isFluxFunction(function.getFactoryMethod())).isTrue();
+		assertThat(FunctionFactoryUtils.isFluxFunction(function.getFactoryMethod())).isTrue();
 		assertThat(function.apply(Flux.just("foo")).blockFirst()).isEqualTo("FOO");
 	}
 

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

@@ -52,7 +52,7 @@ 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.cloud.function.support.FunctionFactoryUtils;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.ResolvableType;
@@ -297,17 +297,17 @@ public class ContextFunctionCatalogAutoConfiguration {
 
 		private boolean isFluxFunction(String name, Function<?, ?> function) {
 			boolean fluxTypes = this.hasFluxTypes(function);
-			return fluxTypes || FunctionUtils.isFluxFunction(function);
+			return fluxTypes || FunctionFactoryUtils.isFluxFunction(function);
 		}
 
 		private boolean isFluxConsumer(String name, Consumer<?> consumer) {
 			boolean fluxTypes = this.hasFluxTypes(consumer);
-			return fluxTypes || FunctionUtils.isFluxConsumer(consumer);
+			return fluxTypes || FunctionFactoryUtils.isFluxConsumer(consumer);
 		}
 
 		private boolean isFluxSupplier(String name, Supplier<?> supplier) {
 			boolean fluxTypes = this.hasFluxTypes(supplier);
-			return fluxTypes || FunctionUtils.isFluxSupplier(supplier);
+			return fluxTypes || FunctionFactoryUtils.isFluxSupplier(supplier);
 		}
 
 		private boolean hasFluxTypes(Object function) {
@@ -379,7 +379,7 @@ public class ContextFunctionCatalogAutoConfiguration {
 				else if (registry instanceof BeanFactory) {
 					Object bean = ((BeanFactory) registry).getBean(name);
 					if (bean instanceof FunctionFactoryMetadata) {
-						FunctionFactoryMetadata factory = (FunctionFactoryMetadata) bean;
+						FunctionFactoryMetadata<?> factory = (FunctionFactoryMetadata<?>) bean;
 						Type type = factory.getFactoryMethod().getGenericReturnType();
 						param = extractType(type, paramType, index);
 					}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2015 the original author or authors.
+ * Copyright 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.

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

@@ -1,34 +0,0 @@
-/*
- * Copyright 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.util.function.Consumer;
-
-/**
- * @author Mark Fisher
- *
- * @param <T> output type of target Consumer
- */
-public interface ConsumerProxy<T> extends Consumer<T>, FunctionFactoryMetadata {
-
-	default boolean isFluxConsumer() {
-		return FunctionUtils.isFluxConsumer(getFactoryMethod());
-	}
-
-	Consumer<T> getTarget();
-
-}

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

@@ -23,8 +23,10 @@ import java.lang.reflect.Method;
  *
  * @param <T>
  */
-public interface FunctionFactoryMetadata {
+public interface FunctionFactoryMetadata<F> {
 
 	Method getFactoryMethod();
 
+	F getTarget();
+
 }

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

@@ -0,0 +1,135 @@
+/*
+ * Copyright 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.invoke.SerializedLambda;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.util.ObjectUtils;
+import org.springframework.util.ReflectionUtils;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * <p>
+ * Miscellaneous utility operations to interrogate functional
+ * components (beans) configured in BeanFactory.
+ * </p>
+ * <p>
+ * It is important to understand that it is not a general purpose utility
+ * to interrogate "any" functional component. Certain operations may/will
+ * not work as expected due to java type erasure. While BeanFactory is not
+ * the requirement, this utility is targeting only the components
+ * defined in such way where they could be configured as
+ * @{@link Component} or @{@link Bean} within BeanFactory.
+ * </p>
+ * It is primarily used internally by the framework.
+ *
+ * @author Mark Fisher
+ * @author Oleg Zhurakousky
+ */
+public abstract class FunctionFactoryUtils {
+
+	private static final String FLUX_CLASS_NAME = Flux.class.getName();
+
+	private FunctionFactoryUtils() {
+	}
+
+	public static boolean isFluxConsumer(Consumer<?> consumer) {
+		return consumer instanceof FunctionFactoryMetadata
+				? isFluxConsumer(((FunctionFactoryMetadata<?>) consumer).getFactoryMethod())
+				: isFlux(1, getParameterizedTypeNames(consumer, Consumer.class));
+	}
+
+	public static boolean isFluxConsumer(Method method) {
+		return isFlux(1, getParameterizedTypeNamesForMethod(method, Consumer.class));
+	}
+
+	public static boolean isFluxSupplier(Supplier<?> supplier) {
+		return supplier instanceof FunctionFactoryMetadata
+				? isFluxSupplier(((FunctionFactoryMetadata<?>) supplier).getFactoryMethod())
+				: isFlux(1, getParameterizedTypeNames(supplier, Supplier.class));
+	}
+
+	public static boolean isFluxSupplier(Method method) {
+		return isFlux(1, getParameterizedTypeNamesForMethod(method, Supplier.class));
+	}
+
+	public static boolean isFluxFunction(Function<?, ?> function) {
+		return function instanceof FunctionFactoryMetadata
+				? isFluxFunction(((FunctionFactoryMetadata<?>) function).getFactoryMethod())
+				: isFlux(1, getParameterizedTypeNames(function, Function.class));
+	}
+
+	public static boolean isFluxFunction(Method method) {
+		return isFlux(2, getParameterizedTypeNamesForMethod(method, Function.class));
+	}
+
+	private static String[] getParameterizedTypeNamesForMethod(Method method, Class<?> interfaceClass) {
+		String[] types = retrieveTypes(method.getGenericReturnType(), interfaceClass);
+		return types == null ? new String[0] : types;
+	}
+
+	private static String[] getParameterizedTypeNames(Object source, Class<?> interfaceClass) {
+		return Stream.of(source.getClass().getGenericInterfaces())
+				.map(gi -> retrieveTypes(gi, interfaceClass))
+				.filter(s -> s != null)
+				.findFirst()
+				.orElse(getSerializedLambdaParameterizedTypeNames(source));
+	}
+
+	private static String[] retrieveTypes(Type genericInterface, Class<?> interfaceClass){
+		if ((genericInterface instanceof ParameterizedType) && interfaceClass
+				.getTypeName().equals(((ParameterizedType) genericInterface).getRawType().getTypeName())) {
+			ParameterizedType type = (ParameterizedType) genericInterface;
+			Type[] args = type.getActualTypeArguments();
+			if (args != null) {
+				return Stream.of(args).map(arg -> arg.getTypeName()).toArray(String[]::new);
+			}
+		}
+		return null;
+	}
+
+	private static String[] getSerializedLambdaParameterizedTypeNames(Object source) {
+		Method method = ReflectionUtils.findMethod(source.getClass(), "writeReplace");
+		if (method == null) {
+			return null;
+		}
+		ReflectionUtils.makeAccessible(method);
+		SerializedLambda serializedLambda = (SerializedLambda) ReflectionUtils
+				.invokeMethod(method, source);
+		String signature = serializedLambda.getImplMethodSignature().replaceAll("[()]","");
+
+		List<String> typeNames = Stream.of(signature.split(";"))
+				.map(t -> t.substring(1).replace('/', '.')).collect(Collectors.toList());
+
+		return typeNames.toArray(new String[typeNames.size()]);
+	}
+
+	private static boolean isFlux(int length, String... types){
+		return !ObjectUtils.isEmpty(types) && types.length == length && Stream.of(types).allMatch(type -> type.startsWith(FLUX_CLASS_NAME));
+	}
+}

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

@@ -1,35 +0,0 @@
-/*
- * Copyright 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.util.function.Function;
-
-/**
- * @author Mark Fisher
- *
- * @param <T> input type of target Function
- * @param <R> output type of target Function
- */
-public interface FunctionProxy<T, R> extends Function<T, R>, FunctionFactoryMetadata {
-
-	default boolean isFluxFunction() {
-		return FunctionUtils.isFluxFunction(getFactoryMethod());
-	}
-
-	Function<T, R> getTarget();
-
-}

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

@@ -1,168 +0,0 @@
-/*
- * Copyright 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.invoke.SerializedLambda;
-import java.lang.reflect.Method;
-import java.lang.reflect.ParameterizedType;
-import java.lang.reflect.Type;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.function.Supplier;
-
-import org.springframework.util.ObjectUtils;
-import org.springframework.util.ReflectionUtils;
-
-import reactor.core.publisher.Flux;
-
-/**
- * @author Mark Fisher
- */
-public abstract class FunctionUtils {
-
-	private static final String FLUX_CLASS_NAME = Flux.class.getName();
-
-	private FunctionUtils() {
-	}
-
-	@SuppressWarnings("rawtypes")
-	public static boolean isFluxConsumer(Consumer<?> consumer) {
-		if (consumer instanceof ConsumerProxy) {
-			return ((ConsumerProxy) consumer).isFluxConsumer();
-		}
-		String[] types = getParameterizedTypeNames(consumer, Consumer.class);
-		if (ObjectUtils.isEmpty(types)) {
-			return false;
-		}
-		return (types[0].startsWith(FLUX_CLASS_NAME));
-	}
-
-	@SuppressWarnings("rawtypes")
-	public static boolean isFluxSupplier(Supplier<?> supplier) {
-		if (supplier instanceof SupplierProxy) {
-			return ((SupplierProxy) supplier).isFluxSupplier();
-		}
-		String[] types = getParameterizedTypeNames(supplier, Supplier.class);
-		if (ObjectUtils.isEmpty(types)) {
-			// 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));
-	}
-
-	@SuppressWarnings("rawtypes")
-	public static boolean isFluxFunction(Function<?, ?> function) {
-		if (function instanceof FunctionProxy) {
-			return ((FunctionProxy) function).isFluxFunction();
-		}
-		String[] types = getParameterizedTypeNames(function, Function.class);
-		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[] getParameterizedTypeNames(Object source,
-			Class<?> interfaceClass) {
-		Type[] genericInterfaces = source.getClass().getGenericInterfaces();
-		for (Type genericInterface : genericInterfaces) {
-			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 getSerializedLambdaParameterizedTypeNames(source);
-	}
-
-	private static String[] getSerializedLambdaParameterizedTypeNames(Object source) {
-		Method method = ReflectionUtils.findMethod(source.getClass(), "writeReplace");
-		if (method == null) {
-			return null;
-		}
-		ReflectionUtils.makeAccessible(method);
-		SerializedLambda serializedLambda = (SerializedLambda) ReflectionUtils
-				.invokeMethod(method, source);
-		String signature = serializedLambda.getImplMethodSignature().replaceAll("[()]",
-				"");
-		List<String> typeNames = new ArrayList<>();
-		for (String types : signature.split(";")) {
-			typeNames.add(types.substring(1).replace('/', '.'));
-		}
-		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

@@ -1,33 +0,0 @@
-/*
- * Copyright 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.util.function.Supplier;
-
-/**
- * @author Mark Fisher
- *
- * @param <T> output type of target Supplier
- */
-public interface SupplierProxy<T> extends Supplier<T>, FunctionFactoryMetadata {
-
-	default boolean isFluxSupplier() {
-		return FunctionUtils.isFluxSupplier(getFactoryMethod());
-	}
-
-	Supplier<T> getTarget();
-}

spring-cloud-function-core/src/test/java/org/springframework/cloud/function/support/FunctionFactoryUtilsTests.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.function.gateway;
+package org.springframework.cloud.function.support;
 
 import java.lang.reflect.Method;
 import java.util.function.Consumer;
@@ -23,7 +23,7 @@ import java.util.function.Supplier;
 
 import org.junit.Test;
 
-import org.springframework.cloud.function.support.FunctionUtils;
+import org.springframework.cloud.function.support.FunctionFactoryUtils;
 import org.springframework.util.ReflectionUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -34,30 +34,30 @@ import reactor.core.publisher.Flux;
  * @author Dave Syer
  *
  */
-public class FunctionUtilsTests {
+public class FunctionFactoryUtilsTests {
 
 	@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();
+		Method method = ReflectionUtils.findMethod(FunctionFactoryUtilsTests.class, "fluxConsumer");
+		assertThat(FunctionFactoryUtils.isFluxConsumer(method)).isTrue();
+		assertThat(FunctionFactoryUtils.isFluxSupplier(method)).isFalse();
+		assertThat(FunctionFactoryUtils.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();
+		Method method = ReflectionUtils.findMethod(FunctionFactoryUtilsTests.class, "fluxSupplier");
+		assertThat(FunctionFactoryUtils.isFluxSupplier(method)).isTrue();
+		assertThat(FunctionFactoryUtils.isFluxConsumer(method)).isFalse();
+		assertThat(FunctionFactoryUtils.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();
+		Method method = ReflectionUtils.findMethod(FunctionFactoryUtilsTests.class, "fluxFunction");
+		assertThat(FunctionFactoryUtils.isFluxFunction(method)).isTrue();
+		assertThat(FunctionFactoryUtils.isFluxSupplier(method)).isFalse();
+		assertThat(FunctionFactoryUtils.isFluxConsumer(method)).isFalse();
 	}
 	
 	public Function<Flux<Foo>, Flux<Foo>> fluxFunction() {