Add FunctionRegistry interface (writable FunctionCatalog)

This makes dynamic function registration (after context starts) much
easier. Also frees us from having to employ BeanFactoryPostProcessor
and other tricks to get the functions registered on startup.

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

@@ -33,19 +33,14 @@ import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
-import java.util.stream.Stream;
 
-import org.springframework.beans.BeansException;
 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;
-import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.config.ConstructorArgumentValues.ValueHolder;
 import org.springframework.beans.factory.support.AbstractBeanDefinition;
-import org.springframework.beans.factory.support.BeanDefinitionRegistry;
-import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.cloud.function.core.FluxConsumer;
@@ -95,83 +90,56 @@ public class ContextFunctionCatalogAutoConfiguration {
 	private Map<String, FunctionRegistration<?>> registrations = Collections.emptyMap();
 
 	@Bean
-	public FunctionCatalog functionCatalog(ContextFunctionPostProcessor processor) {
-		return new BeanFactoryFunctionCatalog(
-				new InMemoryFunctionCatalog(
-						processor.merge(registrations, consumers, suppliers, functions)),
-				processor);
+	public FunctionRegistry functionCatalog(ContextFunctionRegistry processor) {
+		processor.merge(registrations, consumers, suppliers, functions);
+		return new BeanFactoryFunctionCatalog(processor);
 	}
 
 	@Bean
-	public FunctionInspector functionInspector(ContextFunctionPostProcessor processor) {
+	public FunctionInspector functionInspector(ContextFunctionRegistry processor) {
 		return new BeanFactoryFunctionInspector(processor);
 	}
 
-	protected class BeanFactoryFunctionCatalog implements FunctionCatalog {
+	protected static class BeanFactoryFunctionCatalog implements FunctionRegistry {
 
-		private final FunctionCatalog delegate;
-		private final ContextFunctionPostProcessor processor;
+		private final ContextFunctionRegistry processor;
+
+		@Override
+		public <T> void register(FunctionRegistration<T> registration) {
+			processor.register(registration);
+		}
 
 		@SuppressWarnings("unchecked")
 		public <T> Supplier<T> lookupSupplier(String name) {
-			Supplier<T> result = this.delegate.lookupSupplier(name);
-			if (result != null) {
-				return result;
-			}
-			if (name.contains(",")) {
-				Object composed = processor.compose(name);
-				if (composed instanceof Supplier) {
-					result = (Supplier<T>) composed;
-				}
-			}
+			Supplier<T> result = (Supplier<T>) processor.lookupSupplier(name);
 			return result;
 		}
 
 		@SuppressWarnings("unchecked")
 		public <T, R> Function<T, R> lookupFunction(String name) {
-			Function<T, R> result = this.delegate.lookupFunction(name);
-			if (result != null) {
-				return result;
-			}
-			if (name.contains(",")) {
-				Object composed = processor.compose(name);
-				if (composed instanceof Function) {
-					result = (Function<T, R>) composed;
-				}
-			}
+			Function<T, R> result = (Function<T, R>) processor.lookupFunction(name);
 			return result;
 		}
 
 		@SuppressWarnings("unchecked")
 		public <T> Consumer<T> lookupConsumer(String name) {
-			Consumer<T> result = this.delegate.lookupConsumer(name);
-			if (result != null) {
-				return result;
-			}
-			if (name.contains(",")) {
-				Object composed = processor.compose(name);
-				if (composed instanceof Consumer) {
-					result = (Consumer<T>) composed;
-				}
-			}
+			Consumer<T> result = (Consumer<T>) processor.lookupConsumer(name);
 			return result;
 		}
 
 		public Set<String> getSupplierNames() {
-			return this.delegate.getSupplierNames();
+			return this.processor.getSuppliers();
 		}
 
 		public Set<String> getFunctionNames() {
-			return this.delegate.getFunctionNames();
+			return this.processor.getFunctions();
 		}
 
 		public Set<String> getConsumerNames() {
-			return this.delegate.getConsumerNames();
+			return this.processor.getConsumers();
 		}
 
-		public BeanFactoryFunctionCatalog(FunctionCatalog delegate,
-				ContextFunctionPostProcessor processor) {
-			this.delegate = delegate;
+		public BeanFactoryFunctionCatalog(ContextFunctionRegistry processor) {
 			this.processor = processor;
 		}
 
@@ -179,9 +147,9 @@ public class ContextFunctionCatalogAutoConfiguration {
 
 	protected class BeanFactoryFunctionInspector implements FunctionInspector {
 
-		private ContextFunctionPostProcessor processor;
+		private ContextFunctionRegistry processor;
 
-		public BeanFactoryFunctionInspector(ContextFunctionPostProcessor processor) {
+		public BeanFactoryFunctionInspector(ContextFunctionRegistry processor) {
 			this.processor = processor;
 		}
 
@@ -223,32 +191,83 @@ public class ContextFunctionCatalogAutoConfiguration {
 	}
 
 	@Component
-	protected static class ContextFunctionPostProcessor
-			implements BeanDefinitionRegistryPostProcessor {
+	protected static class ContextFunctionRegistry {
 
-		private Set<String> suppliers = new HashSet<>();
-		private Set<String> functions = new HashSet<>();
-		private Set<String> consumers = new HashSet<>();
+		private Map<String, Object> suppliers = new HashMap<>();
+		private Map<String, Object> functions = new HashMap<>();
+		private Map<String, Object> consumers = new HashMap<>();
 
-		private BeanDefinitionRegistry registry;
+		@Autowired
+		private ConfigurableListableBeanFactory registry;
 		private ConversionService conversionService;
 		private Map<Object, String> registrations = new HashMap<>();
-		private Map<String, Object> lookup = new HashMap<>();
 		private Map<String, Map<ParamType, Class<?>>> types = new HashMap<>();
 
-		@Override
-		public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) {
-			this.registry = registry;
+		public Set<String> getSuppliers() {
+			return this.suppliers.keySet();
 		}
 
-		public Object compose(String name) {
+		public Set<String> getConsumers() {
+			return this.consumers.keySet();
+		}
+
+		public Set<String> getFunctions() {
+			return this.functions.keySet();
+		}
+
+		public Supplier<?> lookupSupplier(String name) {
+			Object composed = compose(name, this.suppliers, false);
+			if (composed instanceof Supplier) {
+				return (Supplier<?>) composed;
+			}
+			return null;
+		}
+
+		public Consumer<?> lookupConsumer(String name) {
+			Object composed = compose(name, this.consumers, true);
+			if (composed instanceof Consumer) {
+				return (Consumer<?>) composed;
+			}
+			return null;
+		}
+
+		public Function<?, ?> lookupFunction(String name) {
+			Object composed = compose(name, this.functions, true);
+			if (composed instanceof Function) {
+				return (Function<?, ?>) composed;
+			}
+			return null;
+		}
+
+		private Object compose(String name, Map<String, Object> lookup,
+				boolean hasInput) {
 			if (lookup.containsKey(name)) {
 				return lookup.get(name);
 			}
 			String[] stages = StringUtils.tokenizeToStringArray(name, ",");
-			Object function = Stream.of(stages).map(key -> lookup(key)).sequential()
-					.reduce(this::compose).get();
-			lookup.computeIfAbsent(name, key -> function);
+			Object function = lookup(stages[0],
+					!hasInput || stages.length == 1 ? lookup : this.functions);
+			if (function == null) {
+				return null;
+			}
+			Object other = null;
+			for (int i = 1; i < stages.length - 1; i++) {
+				other = lookup(stages[i], this.functions);
+				if (other == null) {
+					return null;
+				}
+				function = compose(function, other);
+			}
+			if (stages.length > 1) {
+				other = lookup(stages[stages.length - 1],
+						hasInput ? lookup : this.functions);
+				if (other == null) {
+					return null;
+				}
+				function = compose(function, other);
+			}
+			final Object value = function;
+			lookup.computeIfAbsent(name, key -> value);
 			Map<ParamType, Class<?>> values = types.computeIfAbsent(name,
 					key -> new HashMap<>());
 			for (ParamType type : ParamType.values()) {
@@ -265,7 +284,7 @@ public class ContextFunctionCatalogAutoConfiguration {
 			return function;
 		}
 
-		private Object lookup(String name) {
+		private Object lookup(String name, Map<String, Object> lookup) {
 			Object result = lookup.get(name);
 			if (result != null) {
 				Map<ParamType, Class<?>> values = types.computeIfAbsent(name,
@@ -302,22 +321,6 @@ public class ContextFunctionCatalogAutoConfiguration {
 			}
 		}
 
-		@Override
-		public void postProcessBeanFactory(ConfigurableListableBeanFactory factory)
-				throws BeansException {
-			for (String name : factory.getBeanDefinitionNames()) {
-				if (isGeneric(factory, name, Supplier.class)) {
-					this.suppliers.add(name);
-				}
-				else if (isGeneric(factory, name, Function.class)) {
-					this.functions.add(name);
-				}
-				else if (isGeneric(factory, name, Consumer.class)) {
-					this.consumers.add(name);
-				}
-			}
-		}
-
 		@SuppressWarnings("unchecked")
 		public <T> void register(FunctionRegistration<T> function) {
 			wrap((FunctionRegistration<Object>) function,
@@ -377,9 +380,8 @@ public class ContextFunctionCatalogAutoConfiguration {
 		}
 
 		private Object convert(Object function, String value) {
-			if (conversionService == null
-					&& registry instanceof ConfigurableListableBeanFactory) {
-				ConversionService conversionService = ((ConfigurableBeanFactory) this.registry)
+			if (conversionService == null && registry != null) {
+				ConversionService conversionService = this.registry
 						.getConversionService();
 				this.conversionService = conversionService != null ? conversionService
 						: new DefaultConversionService();
@@ -406,18 +408,26 @@ public class ContextFunctionCatalogAutoConfiguration {
 			if (target instanceof Supplier) {
 				findType(target, ParamType.OUTPUT);
 				registration.target(target((Supplier<?>) target, key));
+				for (String name : registration.getNames()) {
+					this.suppliers.put(name, (Supplier<?>) registration.getTarget());
+				}
 			}
 			else if (target instanceof Consumer) {
 				findType(target, ParamType.INPUT);
 				registration.target(target((Consumer<?>) target, key));
+				for (String name : registration.getNames()) {
+					this.consumers.put(name, (Consumer<?>) registration.getTarget());
+				}
 			}
 			else if (target instanceof Function) {
 				findType(target, ParamType.INPUT);
 				findType(target, ParamType.OUTPUT);
 				registration.target(target((Function<?, ?>) target, key));
+				for (String name : registration.getNames()) {
+					this.functions.put(name, (Function<?, ?>) registration.getTarget());
+				}
 			}
 			registrations.remove(target);
-			this.lookup.put(key, registration.getTarget());
 			this.registrations.put(registration.getTarget(), key);
 		}
 
@@ -476,32 +486,6 @@ public class ContextFunctionCatalogAutoConfiguration {
 							.isWrapper(findType(function, ParamType.OUTPUT_WRAPPER));
 		}
 
-		private boolean isGeneric(ConfigurableListableBeanFactory factory, String name,
-				Class<?> functionalInterface) {
-			ResolvableType matchingType = null;
-			ResolvableType[] nonMatchingTypes = null;
-			if (functionalInterface.isAssignableFrom(Function.class)) {
-				matchingType = ResolvableType.forClassWithGenerics(Function.class,
-						Flux.class, Flux.class);
-				nonMatchingTypes = new ResolvableType[] {
-						ResolvableType.forClassWithGenerics(Flux.class, String.class),
-						ResolvableType.forClassWithGenerics(Flux.class, String.class) };
-			}
-			else {
-				nonMatchingTypes = new ResolvableType[] {
-						ResolvableType.forClassWithGenerics(Flux.class, String.class) };
-				if (functionalInterface.isAssignableFrom(Consumer.class)) {
-					matchingType = ResolvableType.forClassWithGenerics(Consumer.class,
-							Flux.class);
-				}
-				matchingType = ResolvableType.forClassWithGenerics(Supplier.class,
-						Flux.class);
-			}
-			return factory.isTypeMatch(name, matchingType)
-					&& !factory.isTypeMatch(name, ResolvableType
-							.forClassWithGenerics(functionalInterface, nonMatchingTypes));
-		}
-
 		private Class<?> findType(String name, AbstractBeanDefinition definition,
 				ParamType paramType) {
 			Object source = definition.getSource();

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

@@ -0,0 +1,29 @@
+/*
+ * 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.context;
+
+import org.springframework.cloud.function.core.FunctionCatalog;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public interface FunctionRegistry extends FunctionCatalog {
+
+	<T> void register(FunctionRegistration<T> registration);
+
+}

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

@@ -16,24 +16,23 @@
 
 package org.springframework.cloud.function.context;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
-import java.util.stream.Stream;
 
-import org.springframework.cloud.function.core.FunctionCatalog;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
 
 /**
  * @author Dave Syer
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  */
-public class InMemoryFunctionCatalog implements FunctionCatalog {
+public class InMemoryFunctionCatalog implements FunctionRegistry {
 
 	private final Map<String, Function<?, ?>> functions;
 
@@ -41,24 +40,48 @@ public class InMemoryFunctionCatalog implements FunctionCatalog {
 
 	private final Map<String, Supplier<?>> suppliers;
 
+	public InMemoryFunctionCatalog() {
+		this(Collections.emptySet());
+	}
+	
+	@Autowired(required=false)
 	public InMemoryFunctionCatalog(Set<FunctionRegistration<?>> registrations) {
 		Assert.notNull(registrations, "'registrations' must not be null");
 		this.suppliers = new HashMap<>();
 		this.functions = new HashMap<>();
 		this.consumers = new HashMap<>();
 		registrations.stream().forEach(reg -> reg.getNames().stream().forEach(name -> {
-			if (reg.getTarget() instanceof Consumer){
+			if (reg.getTarget() instanceof Consumer) {
 				consumers.put(name, (Consumer<?>) reg.getTarget());
 			}
-			else if (reg.getTarget() instanceof Function){
+			else if (reg.getTarget() instanceof Function) {
 				functions.put(name, (Function<?, ?>) reg.getTarget());
 			}
-			else if (reg.getTarget() instanceof Supplier){
+			else if (reg.getTarget() instanceof Supplier) {
 				suppliers.put(name, (Supplier<?>) reg.getTarget());
 			}
 		}));
 	}
 
+	@Override
+	public <T> void register(FunctionRegistration<T> registration) {
+		Map<String, ?> values = null;
+		if (registration.getTarget() instanceof Function) {
+			values = this.functions;
+		}
+		else if (registration.getTarget() instanceof Supplier) {
+			values = this.suppliers;
+		}
+		else {
+			values = this.consumers;
+		}
+		@SuppressWarnings("unchecked")
+		Map<String, Object> map = (Map<String, Object>) values;
+		for (String name : registration.getNames()) {
+			map.put(name, registration.getTarget());
+		}
+	}
+
 	@Override
 	@SuppressWarnings("unchecked")
 	public <T> Supplier<T> lookupSupplier(String name) {

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

@@ -0,0 +1,73 @@
+/*
+ * 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.context;
+
+import java.util.function.Function;
+
+import org.junit.Test;
+
+import org.springframework.cloud.function.context.ContextFunctionCatalogAutoConfiguration.BeanFactoryFunctionCatalog;
+import org.springframework.cloud.function.context.ContextFunctionCatalogAutoConfiguration.ContextFunctionRegistry;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class BeanFactoryFunctionCatalogTests {
+
+	private BeanFactoryFunctionCatalog processor = new BeanFactoryFunctionCatalog(
+			new ContextFunctionRegistry());
+
+	@Test
+	public void basicRegistrationFeatures() {
+		processor.register(new FunctionRegistration<>(new Foos()).names("foos"));
+		Function<Flux<Integer>, Flux<String>> foos = processor.lookupFunction("foos");
+		assertThat(foos.apply(Flux.just(2)).blockFirst()).isEqualTo("4");
+	}
+
+	@Test
+	public void compose() {
+		processor.register(new FunctionRegistration<>(new Foos()).names("foos"));
+		processor.register(new FunctionRegistration<>(new Bars()).names("bars"));
+		Function<Flux<Integer>, Flux<String>> foos = processor
+				.lookupFunction("foos,bars");
+		assertThat(foos.apply(Flux.just(2)).blockFirst()).isEqualTo("Hello 4");
+	}
+
+	protected static class Foos implements Function<Integer, String> {
+
+		@Override
+		public String apply(Integer t) {
+			return "" + 2 * t;
+		}
+
+	}
+
+	protected static class Bars implements Function<String, String> {
+
+		@Override
+		public String apply(String t) {
+			return "Hello " + t;
+		}
+
+	}
+
+}

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

@@ -21,7 +21,7 @@ import java.util.function.Function;
 
 import org.junit.Test;
 
-import org.springframework.cloud.function.context.ContextFunctionCatalogAutoConfiguration.ContextFunctionPostProcessor;
+import org.springframework.cloud.function.context.ContextFunctionCatalogAutoConfiguration.ContextFunctionRegistry;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -31,16 +31,18 @@ import reactor.core.publisher.Flux;
  * @author Dave Syer
  *
  */
+// TODO: test all sorts of error conditions (duplicate registrations, incompatible types
+// for functions with the same name, uncomposable combinations)
 public class ContextFunctionPostProcessorTests {
 
-	private ContextFunctionPostProcessor processor = new ContextFunctionPostProcessor();
+	private ContextFunctionRegistry processor = new ContextFunctionRegistry();
 
 	@Test
 	public void basicRegistrationFeatures() {
 		processor.register(new FunctionRegistration<>(new Foos()).names("foos"));
 		@SuppressWarnings("unchecked")
 		Function<Flux<Integer>, Flux<String>> foos = (Function<Flux<Integer>, Flux<String>>) processor
-				.compose("foos");
+				.lookupFunction("foos");
 		assertThat(foos.apply(Flux.just(2)).blockFirst()).isEqualTo("4");
 	}
 
@@ -52,7 +54,7 @@ public class ContextFunctionPostProcessorTests {
 				Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap());
 		@SuppressWarnings("unchecked")
 		Function<Flux<Integer>, Flux<String>> foos = (Function<Flux<Integer>, Flux<String>>) processor
-				.compose("foos");
+				.lookupFunction("foos");
 		assertThat(foos.apply(Flux.just(2)).blockFirst()).isEqualTo("4");
 	}
 
@@ -62,7 +64,7 @@ public class ContextFunctionPostProcessorTests {
 				Collections.emptyMap(), Collections.singletonMap("foos", new Foos()));
 		@SuppressWarnings("unchecked")
 		Function<Flux<Integer>, Flux<String>> foos = (Function<Flux<Integer>, Flux<String>>) processor
-				.compose("foos");
+				.lookupFunction("foos");
 		assertThat(foos.apply(Flux.just(2)).blockFirst()).isEqualTo("4");
 	}
 
@@ -72,7 +74,7 @@ public class ContextFunctionPostProcessorTests {
 		processor.register(new FunctionRegistration<>(new Bars()).names("bars"));
 		@SuppressWarnings("unchecked")
 		Function<Flux<Integer>, Flux<String>> foos = (Function<Flux<Integer>, Flux<String>>) processor
-				.compose("foos,bars");
+				.lookupFunction("foos,bars");
 		assertThat(foos.apply(Flux.just(2)).blockFirst()).isEqualTo("Hello 4");
 	}