SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

2506 adjust code to enable full native/aot support

Browse Source 
Resolves #2506
This commit is contained in:
Oleg Zhurakousky
2022-09-14 12:54:42 +02:00
parent 9a645bc43b
commit 68f052eca1
2 changed files with 54 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
View File

@@ -152,7 +152,22 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
return binder;
}
private <T> Binder<T, ConsumerProperties, ProducerProperties> doGetBinder(String name,
private <T> Binder<T, ConsumerProperties, ProducerProperties> doGetBinder(String name, Class<? extends T> 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 <T> Binder<T, ConsumerProperties, ProducerProperties> doGetBinderConventional(String name,
Class<? extends T> 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<String> defaultCandidateConfigurations = new HashSet<>();
for (Map.Entry<String, BinderConfiguration> 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<String, Object> 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);

68
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -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<BindableFunctionProxyFactory> proxyFactory = new AtomicReference<>();
final Map<String, Boolean> 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);
}
}