Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-9931: Make consumer endpoints dependant on ChannelInitializer

Browse Source 
Fixes: #9931
Issue link: https://github.com/spring-projects/spring-integration/issues/9931

If no message channel bean is declared explicitly, the XML parser for endpoints
register such a candidate into the global `ChannelInitializer`.
This way, the channel is created when this bean is initialized.
However, there are cases when endpoint bean could be called before `ChannelInitializer` initialization.

* Add `consumerEndpointBuilder.addDependsOn(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME);`
into the `AbstractConsumerEndpointParser` to ensure that `ChannelInitializer` bean is initialized
before the called endpoint bean.
* Adjust `ChannelInitializer` logic to use `registerSingleton()` API instead of `registerBeanDefinition()`
since the last one may cause problems with beans cache when this API is called at runtime.

(cherry picked from commit c9e3de8767)
This commit is contained in:
Artem Bilan
2025-03-28 14:35:46 -04:00
committed by Spring Builds
parent 7b35406fae
commit 6fbeec67a5
2 changed files with 33 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
spring-integration-core/src/main/java/org/springframework/integration/config/ChannelInitializer.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2022 the original author or authors. * Copyright 2002-2025 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -25,17 +25,18 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.util.Assert; import org.springframework.util.Assert;
/** /**
* A {@link InitializingBean} implementation that is responsible for creating * An {@link InitializingBean} implementation that is responsible for creating
* channels that are not explicitly defined but identified via the 'input-channel' * channels that are not explicitly defined but identified via the 'input-channel'
* attribute of the corresponding endpoints. * attribute of the corresponding endpoints.
* * <p>
* This bean plays a role of pre-instantiator since it is instantiated and * This bean plays a role of pre-instantiator since it is instantiated and
* initialized as the very first bean of all SI beans using * initialized as the very first bean of all Spring Integration beans using
* {@link org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler}. * {@link org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler}.
* *
* @author Oleg Zhurakousky * @author Oleg Zhurakousky
@@ -48,7 +49,7 @@ public final class ChannelInitializer implements BeanFactoryAware, InitializingB
private static final Log LOGGER = LogFactory.getLog(ChannelInitializer.class); private static final Log LOGGER = LogFactory.getLog(ChannelInitializer.class);
private volatile BeanFactory beanFactory; private volatile DefaultListableBeanFactory beanFactory;
private volatile boolean autoCreate = true; private volatile boolean autoCreate = true;
@@ -61,49 +62,38 @@ public final class ChannelInitializer implements BeanFactoryAware, InitializingB
@Override @Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException { public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory; this.beanFactory = (DefaultListableBeanFactory) beanFactory;
} }
@Override @Override
public void afterPropertiesSet() { public void afterPropertiesSet() {
Assert.notNull(this.beanFactory, "'beanFactory' must not be null"); Assert.notNull(this.beanFactory, "'beanFactory' must not be null");
if (!this.autoCreate) { if (this.autoCreate) {
return;
}
else {
AutoCreateCandidatesCollector channelCandidatesCollector = AutoCreateCandidatesCollector channelCandidatesCollector =
this.beanFactory.getBean(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME, this.beanFactory.getBean(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME,
AutoCreateCandidatesCollector.class); AutoCreateCandidatesCollector.class);
// at this point channelNames are all resolved with placeholders and SpEL // at this point channelNames are all resolved with placeholders and SpEL
Collection<String> channelNames = channelCandidatesCollector.getChannelNames(); Collection<String> channelNames = channelCandidatesCollector.channelNames;
if (channelNames != null) { if (channelNames != null) {
for (String channelName : channelNames) { for (String channelName : channelNames) {
if (!this.beanFactory.containsBean(channelName)) { if (!this.beanFactory.containsBean(channelName)) {
if (LOGGER.isDebugEnabled()) { if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Auto-creating channel '" + channelName + "' as DirectChannel"); LOGGER.debug("Auto-creating channel '" + channelName + "' as DirectChannel");
} }
IntegrationConfigUtils.autoCreateDirectChannel(channelName, DirectChannel channelToRegister = new DirectChannel();
(BeanDefinitionRegistry) this.beanFactory); this.beanFactory.registerSingleton(channelName, channelToRegister);
this.beanFactory.initializeBean(channelToRegister, channelName);
} }
} }
} }
} }
} }
/* /**
* Collects candidate channel names to be auto-created by ChannelInitializer * Collects candidate channel names to be auto-created by {@link ChannelInitializer}.
* @param channelNames the auto-create candidate channel bean names.
*/ */
public static class AutoCreateCandidatesCollector { public record AutoCreateCandidatesCollector(Collection<String> channelNames) {
private final Collection<String> channelNames;
AutoCreateCandidatesCollector(Collection<String> channelNames) {
this.channelNames = channelNames;
}
public Collection<String> getChannelNames() {
return this.channelNames;
}
} }

24
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractConsumerEndpointParser.java
View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2019 the original author or authors. * Copyright 2002-2025 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -29,6 +29,7 @@ import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.ManagedSet; import org.springframework.beans.factory.support.ManagedSet;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser; import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
@@ -148,7 +149,7 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
String inputChannelName = element.getAttribute(inputChannelAttributeName); String inputChannelName = element.getAttribute(inputChannelAttributeName);
if (!parserContext.getRegistry().containsBeanDefinition(inputChannelName)) { if (!parserContext.getRegistry().containsBeanDefinition(inputChannelName)) {
registerChannelForCreation(parserContext, inputChannelName); registerChannelForCreation(parserContext, inputChannelName, builder);
} }
IntegrationNamespaceUtils.checkAndConfigureFixedSubscriberChannel(element, parserContext, inputChannelName, IntegrationNamespaceUtils.checkAndConfigureFixedSubscriberChannel(element, parserContext, inputChannelName,
handlerBeanName); handlerBeanName);
@@ -177,12 +178,17 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
} }
} }
private void registerChannelForCreation(ParserContext parserContext, String inputChannelName) { private void registerChannelForCreation(ParserContext parserContext, String inputChannelName,
if (parserContext.getRegistry() BeanDefinitionBuilder consumerEndpointBuilder) {
.containsBeanDefinition(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME)) {
BeanDefinition channelRegistry = parserContext.getRegistry(). BeanDefinitionRegistry beanDefinitionRegistry = parserContext.getRegistry();
getBeanDefinition(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME);
if (beanDefinitionRegistry.containsBeanDefinition(
IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME)) {
BeanDefinition channelRegistry =
beanDefinitionRegistry.getBeanDefinition(
IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME);
ConstructorArgumentValues caValues = channelRegistry.getConstructorArgumentValues(); ConstructorArgumentValues caValues = channelRegistry.getConstructorArgumentValues();
ValueHolder vh = caValues.getArgumentValue(0, Collection.class); ValueHolder vh = caValues.getArgumentValue(0, Collection.class);
if (vh == null) { //although it should never happen if it does we can fix it if (vh == null) { //although it should never happen if it does we can fix it
@@ -194,11 +200,13 @@ public abstract class AbstractConsumerEndpointParser extends AbstractBeanDefinit
(Collection<String>) caValues.getArgumentValue(0, Collection.class) (Collection<String>) caValues.getArgumentValue(0, Collection.class)
.getValue(); // NOSONAR see comment above .getValue(); // NOSONAR see comment above
channelCandidateNames.add(inputChannelName); // NOSONAR channelCandidateNames.add(inputChannelName); // NOSONAR
consumerEndpointBuilder.addDependsOn(IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME);
} }
else { else {
parserContext.getReaderContext().error("Failed to locate '" + parserContext.getReaderContext().error("Failed to locate '" +
IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME + "'", IntegrationContextUtils.AUTO_CREATE_CHANNEL_CANDIDATES_BEAN_NAME + "'",
parserContext.getRegistry()); beanDefinitionRegistry);
} }
} }