Cleanup and added more tests

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

@@ -61,31 +61,6 @@ public class FunctionType {
 		this.message = messageType();
 	}
 
-	private Type functionType(Type type) {
-		if (Supplier.class.isAssignableFrom(extractClass(type, ParamType.OUTPUT))) {
-			Type product = extractType(type, ParamType.OUTPUT, 0);
-			Class<?> output = extractClass(product, ParamType.OUTPUT);
-			if (FunctionRegistration.class.isAssignableFrom(output)) {
-				type = extractType(product, ParamType.OUTPUT, 0);
-			}
-			else if (Function.class.isAssignableFrom(output)
-					|| Supplier.class.isAssignableFrom(output)
-					|| Consumer.class.isAssignableFrom(output)) {
-				type = product;
-			}
-		}
-		return type;
-	}
-
-	private boolean messageType() {
-		Class<?> inputType = findType(ParamType.INPUT_INNER_WRAPPER);
-		Class<?> outputType = findType(ParamType.OUTPUT_INNER_WRAPPER);
-		return inputType.getName().startsWith(Message.class.getName())
-				|| Message.class.isAssignableFrom(inputType)
-				|| outputType.getName().startsWith(Message.class.getName())
-				|| Message.class.isAssignableFrom(outputType);
-	}
-
 	public Type getType() {
 		return type;
 	}
@@ -179,17 +154,19 @@ public class FunctionType {
 		if (!isWrapper(input) && !isWrapper(output)) {
 			return this;
 		}
-		if (!isWrapper(input) || !isWrapper(output)) {
+		else if (isWrapper(input) && isWrapper(output)) {
+			if (input.isAssignableFrom(getInputWrapper())
+					&& output.isAssignableFrom(getOutputWrapper())) {
+				return this;
+			}
+			return new FunctionType(ResolvableType.forClassWithGenerics(Function.class,
+					wrapper(input, getInputType()), wrapper(output, getOutputType()))
+					.getType());
+		}
+		else {
 			throw new IllegalArgumentException("Both wrapper types must be wrappers in ("
 					+ input + ", " + output + ")");
 		}
-		if (input.isAssignableFrom(getInputWrapper())
-				&& output.isAssignableFrom(getOutputWrapper())) {
-			return this;
-		}
-		return new FunctionType(ResolvableType.forClassWithGenerics(Function.class,
-				wrapper(input, getInputType()), wrapper(output, getOutputType()))
-				.getType());
 	}
 
 	public FunctionType wrap(Class<?> wrapper) {
@@ -212,6 +189,60 @@ public class FunctionType {
 				.getType());
 	}
 
+	@Override
+	public int hashCode() {
+		final int prime = 31;
+		int result = 1;
+		result = prime * result
+				+ ((inputType == null) ? 0 : inputType.toString().hashCode());
+		result = prime * result
+				+ ((inputWrapper == null) ? 0 : inputWrapper.toString().hashCode());
+		result = prime * result + (message ? 1231 : 1237);
+		result = prime * result
+				+ ((outputType == null) ? 0 : outputType.toString().hashCode());
+		result = prime * result
+				+ ((outputWrapper == null) ? 0 : outputWrapper.toString().hashCode());
+		return result;
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (this == obj)
+			return true;
+		if (obj == null)
+			return false;
+		if (getClass() != obj.getClass())
+			return false;
+		FunctionType other = (FunctionType) obj;
+		if (inputType == null) {
+			if (other.inputType != null)
+				return false;
+		}
+		else if (!inputType.toString().equals(other.inputType.toString()))
+			return false;
+		if (inputWrapper == null) {
+			if (other.inputWrapper != null)
+				return false;
+		}
+		else if (!inputWrapper.toString().equals(other.inputWrapper.toString()))
+			return false;
+		if (message != other.message)
+			return false;
+		if (outputType == null) {
+			if (other.outputType != null)
+				return false;
+		}
+		else if (!outputType.toString().equals(other.outputType.toString()))
+			return false;
+		if (outputWrapper == null) {
+			if (other.outputWrapper != null)
+				return false;
+		}
+		else if (!outputWrapper.toString().equals(other.outputWrapper.toString()))
+			return false;
+		return true;
+	}
+
 	private ResolvableType wrapper(Class<?> wrapper, Class<?> type) {
 		return wrap(this, wrapper, type);
 	}
@@ -383,58 +414,29 @@ public class FunctionType {
 		return param;
 	}
 
-	@Override
+	private Type functionType(Type type) {
-	public int hashCode() {
+		if (Supplier.class.isAssignableFrom(extractClass(type, ParamType.OUTPUT))) {
-		final int prime = 31;
+			Type product = extractType(type, ParamType.OUTPUT, 0);
-		int result = 1;
+			Class<?> output = extractClass(product, ParamType.OUTPUT);
-		result = prime * result
+			if (FunctionRegistration.class.isAssignableFrom(output)) {
-				+ ((inputType == null) ? 0 : inputType.toString().hashCode());
+				type = extractType(product, ParamType.OUTPUT, 0);
-		result = prime * result
+			}
-				+ ((inputWrapper == null) ? 0 : inputWrapper.toString().hashCode());
+			else if (Function.class.isAssignableFrom(output)
-		result = prime * result + (message ? 1231 : 1237);
+					|| Supplier.class.isAssignableFrom(output)
-		result = prime * result
+					|| Consumer.class.isAssignableFrom(output)) {
-				+ ((outputType == null) ? 0 : outputType.toString().hashCode());
+				type = product;
-		result = prime * result
+			}
-				+ ((outputWrapper == null) ? 0 : outputWrapper.toString().hashCode());
+		}
-		return result;
+		return type;
 	}
 
-	@Override
+	private boolean messageType() {
-	public boolean equals(Object obj) {
+		Class<?> inputType = findType(ParamType.INPUT_INNER_WRAPPER);
-		if (this == obj)
+		Class<?> outputType = findType(ParamType.OUTPUT_INNER_WRAPPER);
-			return true;
+		return inputType.getName().startsWith(Message.class.getName())
-		if (obj == null)
+				|| Message.class.isAssignableFrom(inputType)
-			return false;
+				|| outputType.getName().startsWith(Message.class.getName())
-		if (getClass() != obj.getClass())
+				|| Message.class.isAssignableFrom(outputType);
-			return false;
-		FunctionType other = (FunctionType) obj;
-		if (inputType == null) {
-			if (other.inputType != null)
-				return false;
-		}
-		else if (!inputType.toString().equals(other.inputType.toString()))
-			return false;
-		if (inputWrapper == null) {
-			if (other.inputWrapper != null)
-				return false;
-		}
-		else if (!inputWrapper.toString().equals(other.inputWrapper.toString()))
-			return false;
-		if (message != other.message)
-			return false;
-		if (outputType == null) {
-			if (other.outputType != null)
-				return false;
-		}
-		else if (!outputType.toString().equals(other.outputType.toString()))
-			return false;
-		if (outputWrapper == null) {
-			if (other.outputWrapper != null)
-				return false;
-		}
-		else if (!outputWrapper.toString().equals(other.outputWrapper.toString()))
-			return false;
-		return true;
 	}
 
 	enum ParamType {

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

@@ -327,6 +327,9 @@ public class ContextFunctionCatalogAutoConfiguration {
 		@SuppressWarnings("unchecked")
 		private Object compose(Object a, Object b) {
 			if (a instanceof Supplier && b instanceof Function) {
+				if (b instanceof FluxConsumer) {
+					throw new UnsupportedOperationException("Composing Supplier and Consumer is not supported at the moment");
+				}
 				Supplier<Object> supplier = (Supplier<Object>) a;
 				Function<Object, Object> function = (Function<Object, Object>) b;
 				return (Supplier<Object>) () -> function.apply(supplier.get());

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

@@ -16,25 +16,25 @@
 
 package org.springframework.cloud.function.context.config;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 import java.net.URLClassLoader;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
 import org.junit.After;
 import org.junit.Test;
-
 import org.springframework.beans.BeanUtils;
 import org.springframework.cloud.function.context.FunctionRegistration;
 import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration.ContextFunctionRegistry;
 import org.springframework.test.util.ReflectionTestUtils;
 import org.springframework.util.ClassUtils;
 
-import static org.assertj.core.api.Assertions.assertThat;
-
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -102,6 +102,24 @@ public class ContextFunctionPostProcessorTests {
 				.containsExactly("foos|bars");
 	}
 
+	@Test
+	public void supplierAndFunction() {
+		processor.register(new FunctionRegistration<Supplier<String>>(() -> "foo", "supplier"));
+		processor.register(new FunctionRegistration<Function<String, String>>((x) -> x.toUpperCase(), "function"));
+		@SuppressWarnings("unchecked")
+		Supplier<Flux<String>> supplier = (Supplier<Flux<String>>) processor.lookupSupplier("supplier|function");
+		assertThat(supplier.get().blockFirst()).isEqualTo("FOO");
+		assertThat(processor.getRegistration(supplier).getNames()).containsExactly("supplier|function");
+	}
+
+	//TODO we should support it at some point since this is really a Runnable
+	@Test(expected=UnsupportedOperationException.class)
+	public void supplierAndConsumer() {
+		processor.register(new FunctionRegistration<Supplier<String>>(() -> "foo", "supplier"));
+		processor.register(new FunctionRegistration<Consumer<String>>(System.out::println, "consumer"));
+		processor.lookupSupplier("supplier|consumer");
+	}
+
 	@Test
 	public void compose() {
 		processor.register(new FunctionRegistration<>(new Foos(), "foos"));