Miscellaneous clean up, refactoring

Simplified FunctionCatalog structure by no longer registering the actual target function since it is available in wrapper anyway.
Cleaned up logic in RequestProcessor

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

@@ -71,8 +71,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 
 	private final Map<String, Object> functions = new ConcurrentHashMap<>();
 
-	private final Map<String, Object> consumers = new ConcurrentHashMap<>();
-
 	private final Map<Object, String> names = new ConcurrentHashMap<>();
 
 	private final Map<String, FunctionType> types = new ConcurrentHashMap<>();
@@ -97,7 +95,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 		if (type == null) {
 			return new HashSet<String>(getSupplierNames()) {
 				{
-					addAll(getConsumerNames());
 					addAll(getFunctionNames());
 				}
 			};
@@ -105,9 +102,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 		if (Supplier.class.isAssignableFrom(type)) {
 			return this.getSupplierNames();
 		}
-		if (Consumer.class.isAssignableFrom(type)) {
-			return this.getConsumerNames();
-		}
 		if (Function.class.isAssignableFrom(type)) {
 			return this.getFunctionNames();
 		}
@@ -130,14 +124,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 		return this.functions.keySet();
 	}
 
-	/**
-	 * Returns the names of available Consumers.
-	 * @return immutable {@link Set} of available {@link Consumer} names.
-	 */
-	public Set<String> getConsumerNames() {
-		return this.consumers.keySet();
-	}
-
 	public boolean hasSuppliers() {
 		return !CollectionUtils.isEmpty(getSupplierNames());
 	}
@@ -146,10 +132,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 		return !CollectionUtils.isEmpty(getFunctionNames());
 	}
 
-	public boolean hasConsumers() {
-		return !CollectionUtils.isEmpty(getConsumerNames());
-	}
-
 	/**
 	 * The size of this catalog, which is the count of all Suppliers,
 	 * Function and Consumers currently registered.
@@ -158,8 +140,7 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 	 */
 	@Override
 	public int size() {
-		return getSupplierNames().size() + getFunctionNames().size()
+		return getSupplierNames().size() + getFunctionNames().size();
-				+ getConsumerNames().size();
 	}
 
 	public FunctionType getFunctionType(String name) {
@@ -194,7 +175,7 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 		if (StringUtils.hasText(functionName)) {
 			FunctionRegistration<?> registration = new FunctionRegistration<Object>(
 					function, functionName);
-			FunctionType functionType = this.findType(registration);
+			FunctionType functionType = this.findType(registration, functionName);
 			return registration.type(functionType.getType());
 		}
 		return null;
@@ -218,12 +199,14 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 	 */
 	protected void register(FunctionRegistration<?> registration, String key) {
 		Object target = registration.getTarget();
-		this.addName(target, key);
+		if (key.equals("uppercase")) {
+			System.out.println();
+		}
 		if (registration.getType() != null) {
 			this.addType(key, registration.getType());
 		}
 		else {
-			FunctionType functionType = findType(registration);
+			FunctionType functionType = findType(registration, key);
 			this.addType(key, functionType);
 			registration.type(functionType.getType());
 		}
@@ -236,12 +219,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 				this.addSupplier(name, (Supplier<?>) registration.getTarget());
 			}
 		}
-		else if (target instanceof Consumer) {
-			type = Consumer.class;
-			for (String name : registration.getNames()) {
-				this.addConsumer(name, (Consumer<?>) registration.getTarget());
-			}
-		}
 		else if (target instanceof Function) {
 			type = Function.class;
 			for (String name : registration.getNames()) {
@@ -258,8 +235,7 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 		}
 	}
 
-	protected FunctionType findType(FunctionRegistration<?> functionRegistration) {
+	protected FunctionType findType(FunctionRegistration<?> functionRegistration, String name) {
-		String name = this.lookupFunctionName(functionRegistration.getTarget());
 		return functionRegistration.getType() != null
 				?  functionRegistration.getType()
 						:  this.getFunctionType(name);
@@ -274,10 +250,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 		this.functions.put(name, function);
 	}
 
-	protected void addConsumer(String name, Consumer<?> consumer) {
-		this.consumers.put(name, consumer);
-	}
-
 	protected void addType(String name, FunctionType functionType) {
 		this.types.computeIfAbsent(name, str -> functionType);
 	}
@@ -339,16 +311,12 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 				composedFunction = composedRegistration.getTarget();
 				this.addType(name, composedRegistration.getType());
 				this.addName(composedFunction, name);
-				if (composedFunction instanceof Function) {
+				if (composedFunction instanceof Function || composedFunction instanceof Consumer) {
 					this.addFunction(name, (Function<?, ?>) composedFunction);
 				}
-				else if (composedFunction instanceof Consumer) {
-					this.addConsumer(name, (Consumer<?>) composedFunction);
-				}
 				else if (composedFunction instanceof Supplier) {
 					this.addSupplier(name, (Supplier<?>) composedFunction);
 				}
-//				this.register(composedRegistration);
 			}
 
 		}
@@ -361,9 +329,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 		if (result == null) {
 			result = this.functions.get(name);
 		}
-		if (result == null) {
-			result = this.consumers.get(name);
-		}
 		if (result == null && !StringUtils.hasText(name)) {
 			if (supplierFound && this.functions.size() == 1) {
 				result = this.functions.values().iterator().next();
@@ -464,9 +429,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 		Object function = null;
 		if (type == null) {
 			function = this.compose(name, this.functions);
-			if (function == null) {
-				function = this.compose(name, this.consumers);
-			}
 			if (function == null) {
 				function = this.compose(name, this.suppliers);
 			}
@@ -483,12 +445,6 @@ public abstract class AbstractComposableFunctionRegistry implements FunctionRegi
 				function = composed;
 			}
 		}
-		else if (Consumer.class.isAssignableFrom(type)) {
-			Object composed = this.compose(name, this.consumers);
-			if (composed != null && Consumer.class.isAssignableFrom(composed.getClass())) {
-				function = composed;
-			}
-		}
 		return function;
 	}
 

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

@@ -40,8 +40,8 @@ public class InMemoryFunctionCatalog extends AbstractComposableFunctionRegistry
 	}
 
 	@Override
-	protected FunctionType findType(FunctionRegistration<?> functionRegistration) {
+	protected FunctionType findType(FunctionRegistration<?> functionRegistration, String name) {
-		FunctionType functionType = super.findType(functionRegistration);
+		FunctionType functionType = super.findType(functionRegistration, name);
 		if (functionType == null) {
 			functionType = new FunctionType(functionRegistration.getTarget().getClass());
 		}

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

@@ -123,11 +123,6 @@ public class ContextFunctionCatalogAutoConfiguration {
 							.publishEvent(new FunctionUnregistrationEvent(this,
 									Function.class, this.getFunctionNames()));
 				}
-				if (this.hasConsumers()) {
-					this.applicationEventPublisher
-							.publishEvent(new FunctionUnregistrationEvent(this,
-									Consumer.class, this.getConsumerNames()));
-				}
 				if (this.hasSuppliers()) {
 					this.applicationEventPublisher
 							.publishEvent(new FunctionUnregistrationEvent(this,
@@ -137,10 +132,9 @@ public class ContextFunctionCatalogAutoConfiguration {
 		}
 
 		@Override
-		protected FunctionType findType(FunctionRegistration<?> functionRegistration) {
+		protected FunctionType findType(FunctionRegistration<?> functionRegistration, String name) {
-			FunctionType functionType = super.findType(functionRegistration);
+			FunctionType functionType = super.findType(functionRegistration, name);
 			if (functionType == null) {
-				String name = this.lookupFunctionName(functionRegistration.getTarget());
 				functionType = functionByNameExist(name)
 						? new FunctionType(functionRegistration.getTarget().getClass())
 						: new FunctionType(

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.function.context.catalog;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+import org.junit.Ignore;
 import org.junit.Test;
 import reactor.core.publisher.Flux;
 
@@ -37,6 +38,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 public class InMemoryFunctionCatalogTests {
 
 	@Test
+	@Ignore // we no longer have a need to register the actual target function as it is contained within wrapper
 	public void testFunctionRegistration() {
 		TestFunction function = new TestFunction();
 		FunctionRegistration<TestFunction> registration = new FunctionRegistration<>(

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

@@ -453,8 +453,8 @@ public class ContextFunctionCatalogAutoConfigurationTests {
 				.lookup(Function.class, "function");
 		assertThat(function.apply(Flux.just("foo")).blockFirst()).isEqualTo("FOO");
 		assertThat(bean).isNotSameAs(function);
-		assertThat(this.inspector.getRegistration(bean)).isNotNull();
+		assertThat(this.inspector.getRegistration(function)).isNotNull();
-		assertThat(this.inspector.getRegistration(bean).getType())
+		assertThat(this.inspector.getRegistration(function).getType())
 				.isEqualTo(this.inspector.getRegistration(function).getType());
 	}
 

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java

@@ -41,7 +41,6 @@ import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.cloud.function.context.message.MessageUtils;
 import org.springframework.cloud.function.core.FluxConsumer;
-import org.springframework.cloud.function.core.FluxWrapper;
 import org.springframework.cloud.function.core.FluxedConsumer;
 import org.springframework.cloud.function.json.JsonMapper;
 import org.springframework.cloud.function.web.util.HeaderUtils;
@@ -273,18 +272,12 @@ public class RequestProcessor {
 	}
 
 	private boolean isInputMultiple(Object handler) {
-		if (handler instanceof FluxWrapper) {
-			handler = ((FluxWrapper<?>) handler).getTarget();
-		}
 		Class<?> type = this.inspector.getInputType(handler);
 		Class<?> wrapper = this.inspector.getInputWrapper(handler);
 		return Collection.class.isAssignableFrom(type) || Flux.class.equals(wrapper);
 	}
 
 	private boolean isOutputSingle(Object handler) {
-		if (handler instanceof FluxWrapper) {
-			handler = ((FluxWrapper<?>) handler).getTarget();
-		}
 		Class<?> type = this.inspector.getOutputType(handler);
 		Class<?> wrapper = this.inspector.getOutputWrapper(handler);
 		if (Stream.class.isAssignableFrom(type)) {
@@ -297,7 +290,6 @@ public class RequestProcessor {
 	}
 
 	private Publisher<?> body(Object handler, ServerWebExchange exchange) {
-
 		ResolvableType elementType = ResolvableType
 				.forClass(this.inspector.getInputType(handler));
 		ResolvableType actualType = elementType;
@@ -388,22 +380,12 @@ public class RequestProcessor {
 		return Mono.from(function.apply(input));
 	}
 
-	private Object getTargetFunction(Object function) {
-		// we need to get the actual un-fluxed function so we can interrogate for types
-		Object target = this.inspector.getRegistration(function).getTarget();
-		if (target instanceof FluxWrapper) {
-			target = ((FluxWrapper<?>) target).getTarget();
-		}
-		return target;
-	}
-
 	private Type getItemType(Object function) {
 		Class<?> inputType = this.inspector.getInputType(function);
 		if (!Collection.class.isAssignableFrom(inputType)) {
 			return inputType;
 		}
-		Type type = this.inspector.getRegistration(this.getTargetFunction(function))
+		Type type = this.inspector.getRegistration(function).getType().getType();
-				.getType().getType();
 		if (type instanceof ParameterizedType) {
 			type = ((ParameterizedType) type).getActualTypeArguments()[0];
 		}