From 219d056801664586e5e3c0aacfb51194c5d819d2 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Mon, 13 Nov 2017 14:13:59 +0000 Subject: [PATCH] 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. --- ...ntextFunctionCatalogAutoConfiguration.java | 210 ++++++++---------- .../function/context/FunctionRegistry.java | 29 +++ .../context/InMemoryFunctionCatalog.java | 37 ++- .../BeanFactoryFunctionCatalogTests.java | 73 ++++++ .../ContextFunctionPostProcessorTests.java | 14 +- .../cloud/function/core/FunctionCatalog.java | 7 +- 6 files changed, 240 insertions(+), 130 deletions(-) create mode 100644 spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionRegistry.java create mode 100644 spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/BeanFactoryFunctionCatalogTests.java diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java index adbfbef01..ea5822699 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/ContextFunctionCatalogAutoConfiguration.java +++ b/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> 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 void register(FunctionRegistration registration) { + processor.register(registration); + } @SuppressWarnings("unchecked") public Supplier lookupSupplier(String name) { - Supplier result = this.delegate.lookupSupplier(name); - if (result != null) { - return result; - } - if (name.contains(",")) { - Object composed = processor.compose(name); - if (composed instanceof Supplier) { - result = (Supplier) composed; - } - } + Supplier result = (Supplier) processor.lookupSupplier(name); return result; } @SuppressWarnings("unchecked") public Function lookupFunction(String name) { - Function result = this.delegate.lookupFunction(name); - if (result != null) { - return result; - } - if (name.contains(",")) { - Object composed = processor.compose(name); - if (composed instanceof Function) { - result = (Function) composed; - } - } + Function result = (Function) processor.lookupFunction(name); return result; } @SuppressWarnings("unchecked") public Consumer lookupConsumer(String name) { - Consumer result = this.delegate.lookupConsumer(name); - if (result != null) { - return result; - } - if (name.contains(",")) { - Object composed = processor.compose(name); - if (composed instanceof Consumer) { - result = (Consumer) composed; - } - } + Consumer result = (Consumer) processor.lookupConsumer(name); return result; } public Set getSupplierNames() { - return this.delegate.getSupplierNames(); + return this.processor.getSuppliers(); } public Set getFunctionNames() { - return this.delegate.getFunctionNames(); + return this.processor.getFunctions(); } public Set 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 suppliers = new HashSet<>(); - private Set functions = new HashSet<>(); - private Set consumers = new HashSet<>(); + private Map suppliers = new HashMap<>(); + private Map functions = new HashMap<>(); + private Map consumers = new HashMap<>(); - private BeanDefinitionRegistry registry; + @Autowired + private ConfigurableListableBeanFactory registry; private ConversionService conversionService; private Map registrations = new HashMap<>(); - private Map lookup = new HashMap<>(); private Map>> types = new HashMap<>(); - @Override - public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) { - this.registry = registry; + public Set getSuppliers() { + return this.suppliers.keySet(); } - public Object compose(String name) { + public Set getConsumers() { + return this.consumers.keySet(); + } + + public Set 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 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> 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 lookup) { Object result = lookup.get(name); if (result != null) { Map> 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 void register(FunctionRegistration function) { wrap((FunctionRegistration) 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(); diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/FunctionRegistry.java new file mode 100644 index 000000000..369dd5eda --- /dev/null +++ b/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 { + + void register(FunctionRegistration registration); + +} diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/InMemoryFunctionCatalog.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/InMemoryFunctionCatalog.java index fac3eeac3..0963c02ba 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/InMemoryFunctionCatalog.java +++ b/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> functions; @@ -41,24 +40,48 @@ public class InMemoryFunctionCatalog implements FunctionCatalog { private final Map> suppliers; + public InMemoryFunctionCatalog() { + this(Collections.emptySet()); + } + + @Autowired(required=false) public InMemoryFunctionCatalog(Set> 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 void register(FunctionRegistration registration) { + Map 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 map = (Map) values; + for (String name : registration.getNames()) { + map.put(name, registration.getTarget()); + } + } + @Override @SuppressWarnings("unchecked") public Supplier lookupSupplier(String name) { diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/BeanFactoryFunctionCatalogTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/BeanFactoryFunctionCatalogTests.java new file mode 100644 index 000000000..e871d130f --- /dev/null +++ b/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> 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> foos = processor + .lookupFunction("foos,bars"); + assertThat(foos.apply(Flux.just(2)).blockFirst()).isEqualTo("Hello 4"); + } + + protected static class Foos implements Function { + + @Override + public String apply(Integer t) { + return "" + 2 * t; + } + + } + + protected static class Bars implements Function { + + @Override + public String apply(String t) { + return "Hello " + t; + } + + } + +} diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/ContextFunctionPostProcessorTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/ContextFunctionPostProcessorTests.java index cdfe6cb6f..4143a80b4 100644 --- a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/ContextFunctionPostProcessorTests.java +++ b/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> foos = (Function, Flux>) 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> foos = (Function, Flux>) 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> foos = (Function, Flux>) 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> foos = (Function, Flux>) processor - .compose("foos,bars"); + .lookupFunction("foos,bars"); assertThat(foos.apply(Flux.just(2)).blockFirst()).isEqualTo("Hello 4"); } diff --git a/spring-cloud-function-core/src/main/java/org/springframework/cloud/function/core/FunctionCatalog.java b/spring-cloud-function-core/src/main/java/org/springframework/cloud/function/core/FunctionCatalog.java index 2d973fb56..1ee67b360 100644 --- a/spring-cloud-function-core/src/main/java/org/springframework/cloud/function/core/FunctionCatalog.java +++ b/spring-cloud-function-core/src/main/java/org/springframework/cloud/function/core/FunctionCatalog.java @@ -16,7 +16,6 @@ package org.springframework.cloud.function.core; -import java.util.Collections; import java.util.Set; import java.util.function.Consumer; import java.util.function.Function; @@ -33,9 +32,9 @@ public interface FunctionCatalog { Consumer lookupConsumer(String name); - default Set getSupplierNames() { return Collections.emptySet(); } + Set getSupplierNames(); - default Set getFunctionNames() { return Collections.emptySet(); } + Set getFunctionNames(); - default Set getConsumerNames() { return Collections.emptySet(); } + Set getConsumerNames(); }