From 68f052eca1ee4129eaea79a80f4befb3894861df Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 14 Sep 2022 12:54:42 +0200 Subject: [PATCH] 2506 adjust code to enable full native/aot support Resolves #2506 --- .../stream/binder/DefaultBinderFactory.java | 27 ++++++-- .../function/FunctionConfiguration.java | 68 +++++++++---------- 2 files changed, 54 insertions(+), 41 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java index f18d11d89..7e0fb7686 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -152,7 +152,22 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl return binder; } - private Binder doGetBinder(String name, + private Binder doGetBinder(String name, Class bindingTargetType) { + if (CollectionUtils.isEmpty(this.binderChildContextInitializers)) { + return this.doGetBinderConventional(name, bindingTargetType); + } + else { + if (!StringUtils.hasText(name) && this.binderChildContextInitializers.size() == 1) { + String configurationName = this.binderChildContextInitializers.keySet().iterator().next(); + return this.getBinderInstance(configurationName); + } + else { + throw new IllegalStateException("Can't determine which binder to use: " + name + "/" + this.binderChildContextInitializers.size()); + } + } + } + + private Binder doGetBinderConventional(String name, Class bindingTargetType) { if (!MessageChannel.class.isAssignableFrom(bindingTargetType) @@ -165,7 +180,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl // Fall back to a default if no argument is provided if (!StringUtils.hasText(name)) { Assert.notEmpty(this.binderConfigurations, - "A default binder has been requested, but there is no binder available"); + "A default binder has been requested, but there is no binder available "); if (!StringUtils.hasText(this.defaultBinder)) { Set defaultCandidateConfigurations = new HashSet<>(); for (Map.Entry binderConfigurationEntry : this.binderConfigurations @@ -256,11 +271,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl if (!this.binderInstanceCache.containsKey(configurationName)) { this.logger.info("Creating binder: " + configurationName); BinderConfiguration binderConfiguration = this.binderConfigurations.get(configurationName); - Assert.state(binderConfiguration != null, "Unknown binder configuration: " + configurationName); - BinderType binderType = this.binderTypeRegistry.get(binderConfiguration.getBinderType()); - Assert.notNull(binderType, "Binder type " + binderConfiguration.getBinderType() + " is not defined"); Map binderProperties = new HashMap<>(); - this.flatten(null, binderConfiguration.getProperties(), binderProperties); ConfigurableApplicationContext binderProducingContext; if (this.binderChildContextInitializers.containsKey(configurationName)) { @@ -270,6 +281,10 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl binderProducingContext.refresh(); } else { + Assert.state(binderConfiguration != null, "Unknown binder configuration: " + configurationName); + BinderType binderType = this.binderTypeRegistry.get(binderConfiguration.getBinderType()); + Assert.notNull(binderType, "Binder type " + binderConfiguration.getBinderType() + " is not defined"); + this.flatten(null, binderConfiguration.getProperties(), binderProperties); this.logger.info("Constructing binder child context for " + configurationName); binderProducingContext = this.initializeBinderContextSimple(configurationName, binderProperties, binderType, binderConfiguration, true); diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index 9ab2b4aaf..841a05fcf 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -44,9 +44,8 @@ import reactor.util.function.Tuples; import org.springframework.beans.BeansException; import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.beans.factory.config.BeanDefinition; -import org.springframework.beans.factory.config.BeanFactoryPostProcessor; -import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.boot.autoconfigure.AutoConfigureAfter; @@ -147,8 +146,8 @@ public class FunctionConfiguration { } @Bean - public BeanFactoryPostProcessor po(Environment environment) { - return new PollableSourceRegistrar(environment); + public SmartInitializingSingleton po(GenericApplicationContext context) { + return new PollableSourceRegistrar(context); } @Bean @@ -804,29 +803,29 @@ public class FunctionConfiguration { } } - private static class PollableSourceRegistrar implements BeanFactoryPostProcessor { + private static class PollableSourceRegistrar implements SmartInitializingSingleton { private final Environment environment; - PollableSourceRegistrar(Environment environment) { - this.environment = environment; + private final GenericApplicationContext context; + + PollableSourceRegistrar(GenericApplicationContext context) { + this.environment = context.getEnvironment(); + this.context = context; } @Override - public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { + public void afterSingletonsInstantiated() { if (StringUtils.hasText(this.environment.getProperty("spring.cloud.stream.pollable-source"))) { String[] sourceNames = this.environment.getProperty("spring.cloud.stream.pollable-source").split(";"); for (String sourceName : sourceNames) { - RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(sourceName); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(1); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(0); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(new StreamFunctionProperties()); final SupportedBindableFeatures supportedBindableFeatures = new SupportedBindableFeatures(); supportedBindableFeatures.setPollable(true); supportedBindableFeatures.setReactive(false); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(supportedBindableFeatures); - ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(sourceName + "_binding", functionBindableProxyDefinition); + + BindableFunctionProxyFactory proxyFactory = + new BindableFunctionProxyFactory(sourceName, 1, 0, new StreamFunctionProperties(), supportedBindableFeatures); + context.registerBean(sourceName + "_binding", BindableFunctionProxyFactory.class, () -> proxyFactory); } } } @@ -868,7 +867,7 @@ public class FunctionConfiguration { if (StringUtils.hasText(streamFunctionProperties.getDefinition())) { String[] functionDefinitions = this.filterEligibleFunctionDefinitions(); for (String functionDefinition : functionDefinitions) { - RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); + FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition); if (function != null) { //Type functionType = function.getFunctionType(); @@ -885,20 +884,23 @@ public class FunctionConfiguration { this.outputCount = this.getOutputCount(function, false); } - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(functionDefinition); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties); - + AtomicReference proxyFactory = new AtomicReference<>(); final Map reactiveFunctions = streamFunctionProperties.getReactive(); final boolean reactiveFn = reactiveFunctions.get(functionDefinition) != null; if (reactiveFn) { final SupportedBindableFeatures supportedBindableFeatures = new SupportedBindableFeatures(); supportedBindableFeatures.setPollable(false); supportedBindableFeatures.setReactive(true); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(supportedBindableFeatures); + + proxyFactory.set(new BindableFunctionProxyFactory(functionDefinition, + this.inputCount, this.outputCount, this.streamFunctionProperties, supportedBindableFeatures)); } - registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition); + else { + proxyFactory.set(new BindableFunctionProxyFactory(functionDefinition, + this.inputCount, this.outputCount, this.streamFunctionProperties)); + } + ((GenericApplicationContext) this.applicationContext).registerBean(functionDefinition + "_binding", + BindableFunctionProxyFactory.class, () -> proxyFactory.get()); } else { logger.warn("The function definition '" + streamFunctionProperties.getDefinition() + @@ -924,12 +926,10 @@ public class FunctionConfiguration { if (sourceFunc == null || //see https://github.com/spring-cloud/spring-cloud-stream/issues/2229 sourceFunc.isSupplier() || (!sourceFunc.getFunctionDefinition().equals(inputBindingName) && applicationContext.containsBean(inputBindingName))) { - RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(inputBindingName); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(1); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(0); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties); - registry.registerBeanDefinition(inputBindingName + "_binding_in", functionBindableProxyDefinition); + + BindableFunctionProxyFactory proxyFactory = new BindableFunctionProxyFactory(inputBindingName, 1, 0, this.streamFunctionProperties); + ((GenericApplicationContext) this.applicationContext).registerBean(inputBindingName + "_binding_in", + BindableFunctionProxyFactory.class, () -> proxyFactory); } } @@ -939,12 +939,10 @@ public class FunctionConfiguration { if (sourceFunc == null || //see https://github.com/spring-cloud/spring-cloud-stream/issues/2229 sourceFunc.isConsumer() || (!sourceFunc.getFunctionDefinition().equals(outputBindingName) && applicationContext.containsBean(outputBindingName))) { - RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(outputBindingName); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(0); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(1); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties); - registry.registerBeanDefinition(outputBindingName + "_binding_out", functionBindableProxyDefinition); + + BindableFunctionProxyFactory proxyFactory = new BindableFunctionProxyFactory(outputBindingName, 0, 1, this.streamFunctionProperties); + ((GenericApplicationContext) this.applicationContext).registerBean(outputBindingName + "_binding_out", + BindableFunctionProxyFactory.class, () -> proxyFactory); } }