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(); }