Extend support for simple types to Consumer and Supplier

The function catalog now always wraps beans that deal with non-flux
generic types.

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

@@ -0,0 +1,33 @@
+/*
+ * 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> {
+
+	default boolean isFluxConsumer() {
+		return FunctionUtils.isFluxConsumer(getTarget());
+	}
+
+	Consumer<T> getTarget();
+}

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

@@ -0,0 +1,43 @@
+/*
+ * 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;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * {@link Consumer} implementation that wraps a target Consumer so that the target's
+ * simple input type will be wrapped as a {@link Flux} instance.
+ *
+ * @author Dave Syer
+ *
+ * @param <T> input type of target consumer
+ */
+public class FluxConsumer<T> implements Consumer<Flux<T>> {
+
+	private final Consumer<T> function;
+
+	public FluxConsumer(Consumer<T> function) {
+		this.function = function;
+	}
+
+	@Override
+	public void accept(Flux<T> input) {
+		input.subscribe(t -> function.accept(t));
+	}
+}

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

@@ -39,6 +39,6 @@ public class FluxFunction<T, R> implements Function<Flux<T>, Flux<R>> {
 
 	@Override
 	public Flux<R> apply(Flux<T> input) {
-		return input.map(i->this.function.apply(i));
+		return input.map(i -> this.function.apply(i));
 	}
 }

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

@@ -22,6 +22,7 @@ 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;
 
@@ -37,7 +38,20 @@ public abstract class FunctionUtils {
 
 	private static final String FLUX_CLASS_NAME = Flux.class.getName();
 
-	private FunctionUtils() {}
+	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) {
@@ -60,14 +74,17 @@ public abstract class FunctionUtils {
 		if (ObjectUtils.isEmpty(types) || types.length != 2) {
 			return false;
 		}
-		return (types[0].startsWith(FLUX_CLASS_NAME) && types[1].startsWith(FLUX_CLASS_NAME));
+		return (types[0].startsWith(FLUX_CLASS_NAME)
+				&& types[1].startsWith(FLUX_CLASS_NAME));
 	}
 
-	private static String[] getParameterizedTypeNames(Object source, Class<?> interfaceClass) {
+	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())) {
+			if ((genericInterface instanceof ParameterizedType) && interfaceClass
+					.getTypeName().equals(((ParameterizedType) genericInterface)
+							.getRawType().getTypeName())) {
 				ParameterizedType type = (ParameterizedType) genericInterface;
 				Type[] args = type.getActualTypeArguments();
 				if (args != null) {
@@ -88,8 +105,10 @@ public abstract class FunctionUtils {
 			return null;
 		}
 		ReflectionUtils.makeAccessible(method);
-		SerializedLambda serializedLambda = (SerializedLambda) ReflectionUtils.invokeMethod(method, source);
-		String signature = serializedLambda.getImplMethodSignature().replaceAll("[()]", "");
+		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('/', '.'));