Create spring-boot-integration module
This commit is contained in:
Andy Wilkinson
committed by
Phillip Webb
Phillip Webb
parent
0d5a141a41
commit
cf7d8332e2
@@ -0,0 +1,385 @@
|
||||
/*
|
||||
* Copyright 2012-2025 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
|
||||
*
|
||||
* https://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.boot.integration.autoconfigure;
|
||||
|
||||
import java.time.Duration;
|
||||
|
||||
import javax.management.MBeanServer;
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import io.rsocket.transport.netty.server.TcpServerTransport;
|
||||
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnThreading;
|
||||
import org.springframework.boot.autoconfigure.condition.SearchStrategy;
|
||||
import org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.jmx.JmxProperties;
|
||||
import org.springframework.boot.autoconfigure.sql.init.OnDatabaseInitializationCondition;
|
||||
import org.springframework.boot.autoconfigure.task.TaskSchedulingAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.thread.Threading;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.boot.context.properties.PropertyMapper;
|
||||
import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException;
|
||||
import org.springframework.boot.jdbc.init.DataSourceScriptDatabaseInitializer;
|
||||
import org.springframework.boot.task.SimpleAsyncTaskSchedulerBuilder;
|
||||
import org.springframework.boot.task.ThreadPoolTaskSchedulerBuilder;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Conditional;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.integration.config.EnableIntegration;
|
||||
import org.springframework.integration.config.EnableIntegrationManagement;
|
||||
import org.springframework.integration.config.IntegrationComponentScanRegistrar;
|
||||
import org.springframework.integration.config.IntegrationManagementConfigurer;
|
||||
import org.springframework.integration.context.IntegrationContextUtils;
|
||||
import org.springframework.integration.jdbc.store.JdbcMessageStore;
|
||||
import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport;
|
||||
import org.springframework.integration.monitor.IntegrationMBeanExporter;
|
||||
import org.springframework.integration.rsocket.ClientRSocketConnector;
|
||||
import org.springframework.integration.rsocket.IntegrationRSocketEndpoint;
|
||||
import org.springframework.integration.rsocket.ServerRSocketConnector;
|
||||
import org.springframework.integration.rsocket.ServerRSocketMessageHandler;
|
||||
import org.springframework.integration.rsocket.outbound.RSocketOutboundGateway;
|
||||
import org.springframework.integration.scheduling.PollerMetadata;
|
||||
import org.springframework.messaging.rsocket.RSocketRequester;
|
||||
import org.springframework.messaging.rsocket.RSocketStrategies;
|
||||
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
|
||||
import org.springframework.scheduling.Trigger;
|
||||
import org.springframework.scheduling.concurrent.SimpleAsyncTaskScheduler;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
import org.springframework.scheduling.support.CronTrigger;
|
||||
import org.springframework.scheduling.support.PeriodicTrigger;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration
|
||||
* Auto-configuration} for Spring Integration.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
* @author Dave Syer
|
||||
* @author Stephane Nicoll
|
||||
* @author Vedran Pavic
|
||||
* @author Madhura Bhave
|
||||
* @author Yong-Hyun Kim
|
||||
* @author Yanming Zhou
|
||||
* @since 1.1.0
|
||||
*/
|
||||
@AutoConfiguration(beforeName = "org.springframework.boot.rsocket.autoconfigure.RSocketMessagingAutoConfiguration",
|
||||
after = { JmxAutoConfiguration.class, TaskSchedulingAutoConfiguration.class },
|
||||
afterName = "org.springframework.boot.jdbc.autoconfigure.DataSourceAutoConfiguration")
|
||||
@ConditionalOnClass(EnableIntegration.class)
|
||||
@EnableConfigurationProperties({ IntegrationProperties.class, JmxProperties.class })
|
||||
public class IntegrationAutoConfiguration {
|
||||
|
||||
@Bean(name = IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME)
|
||||
@ConditionalOnMissingBean(name = IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME)
|
||||
public static org.springframework.integration.context.IntegrationProperties integrationGlobalProperties(
|
||||
IntegrationProperties properties) {
|
||||
org.springframework.integration.context.IntegrationProperties integrationProperties = new org.springframework.integration.context.IntegrationProperties();
|
||||
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
|
||||
map.from(properties.getChannel().isAutoCreate()).to(integrationProperties::setChannelsAutoCreate);
|
||||
map.from(properties.getChannel().getMaxUnicastSubscribers())
|
||||
.to(integrationProperties::setChannelsMaxUnicastSubscribers);
|
||||
map.from(properties.getChannel().getMaxBroadcastSubscribers())
|
||||
.to(integrationProperties::setChannelsMaxBroadcastSubscribers);
|
||||
map.from(properties.getError().isRequireSubscribers())
|
||||
.to(integrationProperties::setErrorChannelRequireSubscribers);
|
||||
map.from(properties.getError().isIgnoreFailures()).to(integrationProperties::setErrorChannelIgnoreFailures);
|
||||
map.from(properties.getEndpoint().isThrowExceptionOnLateReply())
|
||||
.to(integrationProperties::setMessagingTemplateThrowExceptionOnLateReply);
|
||||
map.from(properties.getEndpoint().getDefaultTimeout())
|
||||
.as(Duration::toMillis)
|
||||
.to(integrationProperties::setEndpointsDefaultTimeout);
|
||||
map.from(properties.getEndpoint().getReadOnlyHeaders())
|
||||
.as(StringUtils::toStringArray)
|
||||
.to(integrationProperties::setReadOnlyHeaders);
|
||||
map.from(properties.getEndpoint().getNoAutoStartup())
|
||||
.as(StringUtils::toStringArray)
|
||||
.to(integrationProperties::setNoAutoStartupEndpoints);
|
||||
return integrationProperties;
|
||||
}
|
||||
|
||||
/**
|
||||
* Basic Spring Integration configuration.
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableIntegration
|
||||
protected static class IntegrationConfiguration {
|
||||
|
||||
@Bean(PollerMetadata.DEFAULT_POLLER)
|
||||
@ConditionalOnMissingBean(name = PollerMetadata.DEFAULT_POLLER)
|
||||
public PollerMetadata defaultPollerMetadata(IntegrationProperties integrationProperties,
|
||||
ObjectProvider<PollerMetadataCustomizer> customizers) {
|
||||
IntegrationProperties.Poller poller = integrationProperties.getPoller();
|
||||
MutuallyExclusiveConfigurationPropertiesException.throwIfMultipleNonNullValuesIn((entries) -> {
|
||||
entries.put("spring.integration.poller.cron",
|
||||
StringUtils.hasText(poller.getCron()) ? poller.getCron() : null);
|
||||
entries.put("spring.integration.poller.fixed-delay", poller.getFixedDelay());
|
||||
entries.put("spring.integration.poller.fixed-rate", poller.getFixedRate());
|
||||
});
|
||||
PollerMetadata pollerMetadata = new PollerMetadata();
|
||||
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
|
||||
map.from(poller::getMaxMessagesPerPoll).to(pollerMetadata::setMaxMessagesPerPoll);
|
||||
map.from(poller::getReceiveTimeout).as(Duration::toMillis).to(pollerMetadata::setReceiveTimeout);
|
||||
map.from(poller).as(this::asTrigger).to(pollerMetadata::setTrigger);
|
||||
customizers.orderedStream().forEach((customizer) -> customizer.customize(pollerMetadata));
|
||||
return pollerMetadata;
|
||||
}
|
||||
|
||||
private Trigger asTrigger(IntegrationProperties.Poller poller) {
|
||||
if (StringUtils.hasText(poller.getCron())) {
|
||||
return new CronTrigger(poller.getCron());
|
||||
}
|
||||
if (poller.getFixedDelay() != null) {
|
||||
return createPeriodicTrigger(poller.getFixedDelay(), poller.getInitialDelay(), false);
|
||||
}
|
||||
if (poller.getFixedRate() != null) {
|
||||
return createPeriodicTrigger(poller.getFixedRate(), poller.getInitialDelay(), true);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private Trigger createPeriodicTrigger(Duration period, Duration initialDelay, boolean fixedRate) {
|
||||
PeriodicTrigger trigger = new PeriodicTrigger(period);
|
||||
if (initialDelay != null) {
|
||||
trigger.setInitialDelay(initialDelay);
|
||||
}
|
||||
trigger.setFixedRate(fixedRate);
|
||||
return trigger;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Expose a standard {@link org.springframework.scheduling.TaskScheduler
|
||||
* TaskScheduler} if the user has not enabled task scheduling explicitly. A
|
||||
* {@link SimpleAsyncTaskScheduler} is exposed if the user enables virtual threads via
|
||||
* {@code spring.threads.virtual.enabled=true}, otherwise
|
||||
* {@link ThreadPoolTaskScheduler}.
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@ConditionalOnMissingBean(name = IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)
|
||||
protected static class IntegrationTaskSchedulerConfiguration {
|
||||
|
||||
@Bean(name = IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)
|
||||
@ConditionalOnBean(ThreadPoolTaskSchedulerBuilder.class)
|
||||
@ConditionalOnThreading(Threading.PLATFORM)
|
||||
public ThreadPoolTaskScheduler taskScheduler(ThreadPoolTaskSchedulerBuilder threadPoolTaskSchedulerBuilder) {
|
||||
return threadPoolTaskSchedulerBuilder.build();
|
||||
}
|
||||
|
||||
@Bean(name = IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)
|
||||
@ConditionalOnBean(SimpleAsyncTaskSchedulerBuilder.class)
|
||||
@ConditionalOnThreading(Threading.VIRTUAL)
|
||||
public SimpleAsyncTaskScheduler taskSchedulerVirtualThreads(
|
||||
SimpleAsyncTaskSchedulerBuilder simpleAsyncTaskSchedulerBuilder) {
|
||||
return simpleAsyncTaskSchedulerBuilder.build();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Spring Integration JMX configuration.
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@ConditionalOnClass(EnableIntegrationMBeanExport.class)
|
||||
@ConditionalOnMissingBean(value = IntegrationMBeanExporter.class, search = SearchStrategy.CURRENT)
|
||||
@ConditionalOnBean(MBeanServer.class)
|
||||
@ConditionalOnBooleanProperty("spring.jmx.enabled")
|
||||
protected static class IntegrationJmxConfiguration {
|
||||
|
||||
@Bean
|
||||
public static IntegrationMBeanExporter integrationMbeanExporter(ApplicationContext applicationContext) {
|
||||
return new IntegrationMBeanExporter() {
|
||||
|
||||
@Override
|
||||
public void afterSingletonsInstantiated() {
|
||||
JmxProperties properties = applicationContext.getBean(JmxProperties.class);
|
||||
String defaultDomain = properties.getDefaultDomain();
|
||||
if (StringUtils.hasLength(defaultDomain)) {
|
||||
setDefaultDomain(defaultDomain);
|
||||
}
|
||||
setServer(applicationContext.getBean(properties.getServer(), MBeanServer.class));
|
||||
super.afterSingletonsInstantiated();
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Integration management configuration.
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@ConditionalOnClass(EnableIntegrationManagement.class)
|
||||
@ConditionalOnMissingBean(value = IntegrationManagementConfigurer.class,
|
||||
name = IntegrationManagementConfigurer.MANAGEMENT_CONFIGURER_NAME, search = SearchStrategy.CURRENT)
|
||||
protected static class IntegrationManagementConfiguration {
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableIntegrationManagement(
|
||||
defaultLoggingEnabled = "${spring.integration.management.default-logging-enabled:true}",
|
||||
observationPatterns = "${spring.integration.management.observation-patterns:}")
|
||||
protected static class EnableIntegrationManagementConfiguration {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Integration component scan configuration.
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@ConditionalOnMissingBean(IntegrationComponentScanRegistrar.class)
|
||||
@Import(IntegrationAutoConfigurationScanRegistrar.class)
|
||||
protected static class IntegrationComponentScanConfiguration {
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Integration JDBC configuration.
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@ConditionalOnClass({ JdbcMessageStore.class, DataSourceScriptDatabaseInitializer.class })
|
||||
@ConditionalOnSingleCandidate(DataSource.class)
|
||||
@Conditional(OnIntegrationDatasourceInitializationCondition.class)
|
||||
protected static class IntegrationJdbcConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public IntegrationDataSourceScriptDatabaseInitializer integrationDataSourceInitializer(DataSource dataSource,
|
||||
IntegrationProperties properties) {
|
||||
return new IntegrationDataSourceScriptDatabaseInitializer(dataSource, properties.getJdbc());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* Integration RSocket configuration.
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@ConditionalOnClass({ IntegrationRSocketEndpoint.class, RSocketRequester.class, io.rsocket.RSocket.class })
|
||||
@Conditional(IntegrationRSocketConfiguration.AnyRSocketChannelAdapterAvailable.class)
|
||||
protected static class IntegrationRSocketConfiguration {
|
||||
|
||||
/**
|
||||
* Check if either an {@link IntegrationRSocketEndpoint} or
|
||||
* {@link RSocketOutboundGateway} bean is available.
|
||||
*/
|
||||
static class AnyRSocketChannelAdapterAvailable extends AnyNestedCondition {
|
||||
|
||||
AnyRSocketChannelAdapterAvailable() {
|
||||
super(ConfigurationPhase.REGISTER_BEAN);
|
||||
}
|
||||
|
||||
@ConditionalOnBean(IntegrationRSocketEndpoint.class)
|
||||
static class IntegrationRSocketEndpointAvailable {
|
||||
|
||||
}
|
||||
|
||||
@ConditionalOnBean(RSocketOutboundGateway.class)
|
||||
static class RSocketOutboundGatewayAvailable {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@ConditionalOnClass(TcpServerTransport.class)
|
||||
protected static class IntegrationRSocketServerConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(ServerRSocketMessageHandler.class)
|
||||
public RSocketMessageHandler serverRSocketMessageHandler(RSocketStrategies rSocketStrategies,
|
||||
IntegrationProperties integrationProperties) {
|
||||
|
||||
RSocketMessageHandler messageHandler = new ServerRSocketMessageHandler(
|
||||
integrationProperties.getRsocket().getServer().isMessageMappingEnabled());
|
||||
messageHandler.setRSocketStrategies(rSocketStrategies);
|
||||
return messageHandler;
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public ServerRSocketConnector serverRSocketConnector(ServerRSocketMessageHandler messageHandler) {
|
||||
return new ServerRSocketConnector(messageHandler);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
protected static class IntegrationRSocketClientConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
@Conditional(RemoteRSocketServerAddressConfigured.class)
|
||||
public ClientRSocketConnector clientRSocketConnector(IntegrationProperties integrationProperties,
|
||||
RSocketStrategies rSocketStrategies) {
|
||||
|
||||
IntegrationProperties.RSocket.Client client = integrationProperties.getRsocket().getClient();
|
||||
ClientRSocketConnector clientRSocketConnector = (client.getUri() != null)
|
||||
? new ClientRSocketConnector(client.getUri())
|
||||
: new ClientRSocketConnector(client.getHost(), client.getPort());
|
||||
clientRSocketConnector.setRSocketStrategies(rSocketStrategies);
|
||||
return clientRSocketConnector;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if a remote address is configured for the RSocket Integration client.
|
||||
*/
|
||||
static class RemoteRSocketServerAddressConfigured extends AnyNestedCondition {
|
||||
|
||||
RemoteRSocketServerAddressConfigured() {
|
||||
super(ConfigurationPhase.REGISTER_BEAN);
|
||||
}
|
||||
|
||||
@ConditionalOnProperty("spring.integration.rsocket.client.uri")
|
||||
static class WebSocketAddressConfigured {
|
||||
|
||||
}
|
||||
|
||||
@ConditionalOnProperty({ "spring.integration.rsocket.client.host",
|
||||
"spring.integration.rsocket.client.port" })
|
||||
static class TcpAddressConfigured {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
static class OnIntegrationDatasourceInitializationCondition extends OnDatabaseInitializationCondition {
|
||||
|
||||
OnIntegrationDatasourceInitializationCondition() {
|
||||
super("Integration", "spring.integration.jdbc.initialize-schema");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
/*
|
||||
* Copyright 2012-2024 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
|
||||
*
|
||||
* https://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.boot.integration.autoconfigure;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
|
||||
import org.springframework.beans.BeansException;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.BeanFactoryAware;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigurationPackages;
|
||||
import org.springframework.core.annotation.AnnotationAttributes;
|
||||
import org.springframework.core.type.AnnotationMetadata;
|
||||
import org.springframework.integration.annotation.IntegrationComponentScan;
|
||||
import org.springframework.integration.config.IntegrationComponentScanRegistrar;
|
||||
|
||||
/**
|
||||
* Variation of {@link IntegrationComponentScanRegistrar} the links
|
||||
* {@link AutoConfigurationPackages}.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
* @author Phillip Webb
|
||||
*/
|
||||
class IntegrationAutoConfigurationScanRegistrar extends IntegrationComponentScanRegistrar implements BeanFactoryAware {
|
||||
|
||||
private BeanFactory beanFactory;
|
||||
|
||||
@Override
|
||||
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
|
||||
this.beanFactory = beanFactory;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata,
|
||||
final BeanDefinitionRegistry registry) {
|
||||
super.registerBeanDefinitions(AnnotationMetadata.introspect(IntegrationComponentScanConfiguration.class),
|
||||
registry);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Collection<String> getBasePackages(AnnotationAttributes componentScan, BeanDefinitionRegistry registry) {
|
||||
return (AutoConfigurationPackages.has(this.beanFactory) ? AutoConfigurationPackages.get(this.beanFactory)
|
||||
: Collections.emptyList());
|
||||
}
|
||||
|
||||
@IntegrationComponentScan
|
||||
private static final class IntegrationComponentScanConfiguration {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
/*
|
||||
* Copyright 2012-2022 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
|
||||
*
|
||||
* https://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.boot.integration.autoconfigure;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.springframework.boot.jdbc.DatabaseDriver;
|
||||
import org.springframework.boot.jdbc.init.DataSourceScriptDatabaseInitializer;
|
||||
import org.springframework.boot.jdbc.init.PlatformPlaceholderDatabaseDriverResolver;
|
||||
import org.springframework.boot.sql.init.DatabaseInitializationSettings;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* {@link DataSourceScriptDatabaseInitializer} for the Spring Integration database. May be
|
||||
* registered as a bean to override auto-configuration.
|
||||
*
|
||||
* @author Vedran Pavic
|
||||
* @author Andy Wilkinson
|
||||
* @since 2.6.0
|
||||
*/
|
||||
public class IntegrationDataSourceScriptDatabaseInitializer extends DataSourceScriptDatabaseInitializer {
|
||||
|
||||
/**
|
||||
* Create a new {@link IntegrationDataSourceScriptDatabaseInitializer} instance.
|
||||
* @param dataSource the Spring Integration data source
|
||||
* @param properties the Spring Integration JDBC properties
|
||||
* @see #getSettings
|
||||
*/
|
||||
public IntegrationDataSourceScriptDatabaseInitializer(DataSource dataSource,
|
||||
IntegrationProperties.Jdbc properties) {
|
||||
this(dataSource, getSettings(dataSource, properties));
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new {@link IntegrationDataSourceScriptDatabaseInitializer} instance.
|
||||
* @param dataSource the Spring Integration data source
|
||||
* @param settings the database initialization settings
|
||||
* @see #getSettings
|
||||
*/
|
||||
public IntegrationDataSourceScriptDatabaseInitializer(DataSource dataSource,
|
||||
DatabaseInitializationSettings settings) {
|
||||
super(dataSource, settings);
|
||||
}
|
||||
|
||||
/**
|
||||
* Adapts {@link IntegrationProperties.Jdbc Spring Integration JDBC properties} to
|
||||
* {@link DatabaseInitializationSettings} replacing any {@literal @@platform@@}
|
||||
* placeholders.
|
||||
* @param dataSource the Spring Integration data source
|
||||
* @param properties the Spring Integration JDBC properties
|
||||
* @return a new {@link DatabaseInitializationSettings} instance
|
||||
* @see #IntegrationDataSourceScriptDatabaseInitializer(DataSource,
|
||||
* DatabaseInitializationSettings)
|
||||
*/
|
||||
static DatabaseInitializationSettings getSettings(DataSource dataSource, IntegrationProperties.Jdbc properties) {
|
||||
DatabaseInitializationSettings settings = new DatabaseInitializationSettings();
|
||||
settings.setSchemaLocations(resolveSchemaLocations(dataSource, properties));
|
||||
settings.setMode(properties.getInitializeSchema());
|
||||
settings.setContinueOnError(true);
|
||||
return settings;
|
||||
}
|
||||
|
||||
private static List<String> resolveSchemaLocations(DataSource dataSource, IntegrationProperties.Jdbc properties) {
|
||||
PlatformPlaceholderDatabaseDriverResolver platformResolver = new PlatformPlaceholderDatabaseDriverResolver();
|
||||
platformResolver = platformResolver.withDriverPlatform(DatabaseDriver.MARIADB, "mysql");
|
||||
if (StringUtils.hasText(properties.getPlatform())) {
|
||||
return platformResolver.resolveAll(properties.getPlatform(), properties.getSchema());
|
||||
}
|
||||
return platformResolver.resolveAll(dataSource, properties.getSchema());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,458 @@
|
||||
/*
|
||||
* Copyright 2012-2025 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
|
||||
*
|
||||
* https://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.boot.integration.autoconfigure;
|
||||
|
||||
import java.net.URI;
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.boot.sql.init.DatabaseInitializationMode;
|
||||
|
||||
/**
|
||||
* Configuration properties for Spring Integration.
|
||||
*
|
||||
* @author Vedran Pavic
|
||||
* @author Stephane Nicoll
|
||||
* @author Artem Bilan
|
||||
* @since 2.0.0
|
||||
*/
|
||||
@ConfigurationProperties("spring.integration")
|
||||
public class IntegrationProperties {
|
||||
|
||||
private final Channel channel = new Channel();
|
||||
|
||||
private final Endpoint endpoint = new Endpoint();
|
||||
|
||||
private final Error error = new Error();
|
||||
|
||||
private final Jdbc jdbc = new Jdbc();
|
||||
|
||||
private final RSocket rsocket = new RSocket();
|
||||
|
||||
private final Poller poller = new Poller();
|
||||
|
||||
private final Management management = new Management();
|
||||
|
||||
public Channel getChannel() {
|
||||
return this.channel;
|
||||
}
|
||||
|
||||
public Endpoint getEndpoint() {
|
||||
return this.endpoint;
|
||||
}
|
||||
|
||||
public Error getError() {
|
||||
return this.error;
|
||||
}
|
||||
|
||||
public Jdbc getJdbc() {
|
||||
return this.jdbc;
|
||||
}
|
||||
|
||||
public RSocket getRsocket() {
|
||||
return this.rsocket;
|
||||
}
|
||||
|
||||
public Poller getPoller() {
|
||||
return this.poller;
|
||||
}
|
||||
|
||||
public Management getManagement() {
|
||||
return this.management;
|
||||
}
|
||||
|
||||
public static class Channel {
|
||||
|
||||
/**
|
||||
* Whether to create input channels if necessary.
|
||||
*/
|
||||
private boolean autoCreate = true;
|
||||
|
||||
/**
|
||||
* Default number of subscribers allowed on, for example, a 'DirectChannel'.
|
||||
*/
|
||||
private int maxUnicastSubscribers = Integer.MAX_VALUE;
|
||||
|
||||
/**
|
||||
* Default number of subscribers allowed on, for example, a
|
||||
* 'PublishSubscribeChannel'.
|
||||
*/
|
||||
private int maxBroadcastSubscribers = Integer.MAX_VALUE;
|
||||
|
||||
public void setAutoCreate(boolean autoCreate) {
|
||||
this.autoCreate = autoCreate;
|
||||
}
|
||||
|
||||
public boolean isAutoCreate() {
|
||||
return this.autoCreate;
|
||||
}
|
||||
|
||||
public void setMaxUnicastSubscribers(int maxUnicastSubscribers) {
|
||||
this.maxUnicastSubscribers = maxUnicastSubscribers;
|
||||
}
|
||||
|
||||
public int getMaxUnicastSubscribers() {
|
||||
return this.maxUnicastSubscribers;
|
||||
}
|
||||
|
||||
public void setMaxBroadcastSubscribers(int maxBroadcastSubscribers) {
|
||||
this.maxBroadcastSubscribers = maxBroadcastSubscribers;
|
||||
}
|
||||
|
||||
public int getMaxBroadcastSubscribers() {
|
||||
return this.maxBroadcastSubscribers;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Endpoint {
|
||||
|
||||
/**
|
||||
* Whether to throw an exception when a reply is not expected anymore by a
|
||||
* gateway.
|
||||
*/
|
||||
private boolean throwExceptionOnLateReply = false;
|
||||
|
||||
/**
|
||||
* List of message header names that should not be populated into Message
|
||||
* instances during a header copying operation.
|
||||
*/
|
||||
private List<String> readOnlyHeaders = new ArrayList<>();
|
||||
|
||||
/**
|
||||
* List of endpoint bean names patterns that should not be started automatically
|
||||
* during application startup.
|
||||
*/
|
||||
private List<String> noAutoStartup = new ArrayList<>();
|
||||
|
||||
/**
|
||||
* Default timeout for blocking operations such as sending or receiving messages.
|
||||
*/
|
||||
private Duration defaultTimeout = Duration.ofSeconds(30);
|
||||
|
||||
public void setThrowExceptionOnLateReply(boolean throwExceptionOnLateReply) {
|
||||
this.throwExceptionOnLateReply = throwExceptionOnLateReply;
|
||||
}
|
||||
|
||||
public boolean isThrowExceptionOnLateReply() {
|
||||
return this.throwExceptionOnLateReply;
|
||||
}
|
||||
|
||||
public List<String> getReadOnlyHeaders() {
|
||||
return this.readOnlyHeaders;
|
||||
}
|
||||
|
||||
public void setReadOnlyHeaders(List<String> readOnlyHeaders) {
|
||||
this.readOnlyHeaders = readOnlyHeaders;
|
||||
}
|
||||
|
||||
public List<String> getNoAutoStartup() {
|
||||
return this.noAutoStartup;
|
||||
}
|
||||
|
||||
public void setNoAutoStartup(List<String> noAutoStartup) {
|
||||
this.noAutoStartup = noAutoStartup;
|
||||
}
|
||||
|
||||
public Duration getDefaultTimeout() {
|
||||
return this.defaultTimeout;
|
||||
}
|
||||
|
||||
public void setDefaultTimeout(Duration defaultTimeout) {
|
||||
this.defaultTimeout = defaultTimeout;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Error {
|
||||
|
||||
/**
|
||||
* Whether to not silently ignore messages on the global 'errorChannel' when there
|
||||
* are no subscribers.
|
||||
*/
|
||||
private boolean requireSubscribers = true;
|
||||
|
||||
/**
|
||||
* Whether to ignore failures for one or more of the handlers of the global
|
||||
* 'errorChannel'.
|
||||
*/
|
||||
private boolean ignoreFailures = true;
|
||||
|
||||
public boolean isRequireSubscribers() {
|
||||
return this.requireSubscribers;
|
||||
}
|
||||
|
||||
public void setRequireSubscribers(boolean requireSubscribers) {
|
||||
this.requireSubscribers = requireSubscribers;
|
||||
}
|
||||
|
||||
public boolean isIgnoreFailures() {
|
||||
return this.ignoreFailures;
|
||||
}
|
||||
|
||||
public void setIgnoreFailures(boolean ignoreFailures) {
|
||||
this.ignoreFailures = ignoreFailures;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Jdbc {
|
||||
|
||||
private static final String DEFAULT_SCHEMA_LOCATION = "classpath:org/springframework/"
|
||||
+ "integration/jdbc/schema-@@platform@@.sql";
|
||||
|
||||
/**
|
||||
* Path to the SQL file to use to initialize the database schema.
|
||||
*/
|
||||
private String schema = DEFAULT_SCHEMA_LOCATION;
|
||||
|
||||
/**
|
||||
* Platform to use in initialization scripts if the @@platform@@ placeholder is
|
||||
* used. Auto-detected by default.
|
||||
*/
|
||||
private String platform;
|
||||
|
||||
/**
|
||||
* Database schema initialization mode.
|
||||
*/
|
||||
private DatabaseInitializationMode initializeSchema = DatabaseInitializationMode.EMBEDDED;
|
||||
|
||||
public String getSchema() {
|
||||
return this.schema;
|
||||
}
|
||||
|
||||
public void setSchema(String schema) {
|
||||
this.schema = schema;
|
||||
}
|
||||
|
||||
public String getPlatform() {
|
||||
return this.platform;
|
||||
}
|
||||
|
||||
public void setPlatform(String platform) {
|
||||
this.platform = platform;
|
||||
}
|
||||
|
||||
public DatabaseInitializationMode getInitializeSchema() {
|
||||
return this.initializeSchema;
|
||||
}
|
||||
|
||||
public void setInitializeSchema(DatabaseInitializationMode initializeSchema) {
|
||||
this.initializeSchema = initializeSchema;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class RSocket {
|
||||
|
||||
private final Client client = new Client();
|
||||
|
||||
private final Server server = new Server();
|
||||
|
||||
public Client getClient() {
|
||||
return this.client;
|
||||
}
|
||||
|
||||
public Server getServer() {
|
||||
return this.server;
|
||||
}
|
||||
|
||||
public static class Client {
|
||||
|
||||
/**
|
||||
* TCP RSocket server host to connect to.
|
||||
*/
|
||||
private String host;
|
||||
|
||||
/**
|
||||
* TCP RSocket server port to connect to.
|
||||
*/
|
||||
private Integer port;
|
||||
|
||||
/**
|
||||
* WebSocket RSocket server uri to connect to.
|
||||
*/
|
||||
private URI uri;
|
||||
|
||||
public void setHost(String host) {
|
||||
this.host = host;
|
||||
}
|
||||
|
||||
public String getHost() {
|
||||
return this.host;
|
||||
}
|
||||
|
||||
public void setPort(Integer port) {
|
||||
this.port = port;
|
||||
}
|
||||
|
||||
public Integer getPort() {
|
||||
return this.port;
|
||||
}
|
||||
|
||||
public void setUri(URI uri) {
|
||||
this.uri = uri;
|
||||
}
|
||||
|
||||
public URI getUri() {
|
||||
return this.uri;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Server {
|
||||
|
||||
/**
|
||||
* Whether to handle message mapping for RSocket through Spring Integration.
|
||||
*/
|
||||
private boolean messageMappingEnabled;
|
||||
|
||||
public boolean isMessageMappingEnabled() {
|
||||
return this.messageMappingEnabled;
|
||||
}
|
||||
|
||||
public void setMessageMappingEnabled(boolean messageMappingEnabled) {
|
||||
this.messageMappingEnabled = messageMappingEnabled;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Poller {
|
||||
|
||||
/**
|
||||
* Maximum number of messages to poll per polling cycle.
|
||||
*/
|
||||
private int maxMessagesPerPoll = Integer.MIN_VALUE; // PollerMetadata.MAX_MESSAGES_UNBOUNDED
|
||||
|
||||
/**
|
||||
* How long to wait for messages on poll.
|
||||
*/
|
||||
private Duration receiveTimeout = Duration.ofSeconds(1); // PollerMetadata.DEFAULT_RECEIVE_TIMEOUT
|
||||
|
||||
/**
|
||||
* Polling delay period. Mutually exclusive with 'cron' and 'fixedRate'.
|
||||
*/
|
||||
private Duration fixedDelay;
|
||||
|
||||
/**
|
||||
* Polling rate period. Mutually exclusive with 'fixedDelay' and 'cron'.
|
||||
*/
|
||||
private Duration fixedRate;
|
||||
|
||||
/**
|
||||
* Polling initial delay. Applied for 'fixedDelay' and 'fixedRate'; ignored for
|
||||
* 'cron'.
|
||||
*/
|
||||
private Duration initialDelay;
|
||||
|
||||
/**
|
||||
* Cron expression for polling. Mutually exclusive with 'fixedDelay' and
|
||||
* 'fixedRate'.
|
||||
*/
|
||||
private String cron;
|
||||
|
||||
public int getMaxMessagesPerPoll() {
|
||||
return this.maxMessagesPerPoll;
|
||||
}
|
||||
|
||||
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
|
||||
this.maxMessagesPerPoll = maxMessagesPerPoll;
|
||||
}
|
||||
|
||||
public Duration getReceiveTimeout() {
|
||||
return this.receiveTimeout;
|
||||
}
|
||||
|
||||
public void setReceiveTimeout(Duration receiveTimeout) {
|
||||
this.receiveTimeout = receiveTimeout;
|
||||
}
|
||||
|
||||
public Duration getFixedDelay() {
|
||||
return this.fixedDelay;
|
||||
}
|
||||
|
||||
public void setFixedDelay(Duration fixedDelay) {
|
||||
this.fixedDelay = fixedDelay;
|
||||
}
|
||||
|
||||
public Duration getFixedRate() {
|
||||
return this.fixedRate;
|
||||
}
|
||||
|
||||
public void setFixedRate(Duration fixedRate) {
|
||||
this.fixedRate = fixedRate;
|
||||
}
|
||||
|
||||
public Duration getInitialDelay() {
|
||||
return this.initialDelay;
|
||||
}
|
||||
|
||||
public void setInitialDelay(Duration initialDelay) {
|
||||
this.initialDelay = initialDelay;
|
||||
}
|
||||
|
||||
public String getCron() {
|
||||
return this.cron;
|
||||
}
|
||||
|
||||
public void setCron(String cron) {
|
||||
this.cron = cron;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class Management {
|
||||
|
||||
/**
|
||||
* Whether Spring Integration components should perform logging in the main
|
||||
* message flow. When disabled, such logging will be skipped without checking the
|
||||
* logging level. When enabled, such logging is controlled as normal by the
|
||||
* logging system's log level configuration.
|
||||
*/
|
||||
private boolean defaultLoggingEnabled = true;
|
||||
|
||||
/**
|
||||
* List of simple patterns to match against the names of Spring Integration
|
||||
* components. When matched, observation instrumentation will be performed for the
|
||||
* component. Please refer to the javadoc of the smartMatch method of Spring
|
||||
* Integration's PatternMatchUtils for details of the pattern syntax.
|
||||
*/
|
||||
private List<String> observationPatterns = new ArrayList<>();
|
||||
|
||||
public boolean isDefaultLoggingEnabled() {
|
||||
return this.defaultLoggingEnabled;
|
||||
}
|
||||
|
||||
public void setDefaultLoggingEnabled(boolean defaultLoggingEnabled) {
|
||||
this.defaultLoggingEnabled = defaultLoggingEnabled;
|
||||
}
|
||||
|
||||
public List<String> getObservationPatterns() {
|
||||
return this.observationPatterns;
|
||||
}
|
||||
|
||||
public void setObservationPatterns(List<String> observationPatterns) {
|
||||
this.observationPatterns = observationPatterns;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
/*
|
||||
* Copyright 2012-2024 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
|
||||
*
|
||||
* https://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.boot.integration.autoconfigure;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.env.EnvironmentPostProcessor;
|
||||
import org.springframework.boot.env.OriginTrackedMapPropertySource;
|
||||
import org.springframework.boot.env.PropertiesPropertySourceLoader;
|
||||
import org.springframework.boot.origin.Origin;
|
||||
import org.springframework.boot.origin.OriginLookup;
|
||||
import org.springframework.core.Ordered;
|
||||
import org.springframework.core.env.ConfigurableEnvironment;
|
||||
import org.springframework.core.env.PropertySource;
|
||||
import org.springframework.core.io.ClassPathResource;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.integration.context.IntegrationProperties;
|
||||
|
||||
/**
|
||||
* An {@link EnvironmentPostProcessor} that maps the configuration of
|
||||
* {@code META-INF/spring.integration.properties} in the environment.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
* @author Stephane Nicoll
|
||||
*/
|
||||
class IntegrationPropertiesEnvironmentPostProcessor implements EnvironmentPostProcessor, Ordered {
|
||||
|
||||
@Override
|
||||
public int getOrder() {
|
||||
return Ordered.LOWEST_PRECEDENCE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) {
|
||||
Resource resource = new ClassPathResource("META-INF/spring.integration.properties");
|
||||
if (resource.exists()) {
|
||||
registerIntegrationPropertiesPropertySource(environment, resource);
|
||||
}
|
||||
}
|
||||
|
||||
protected void registerIntegrationPropertiesPropertySource(ConfigurableEnvironment environment, Resource resource) {
|
||||
PropertiesPropertySourceLoader loader = new PropertiesPropertySourceLoader();
|
||||
try {
|
||||
OriginTrackedMapPropertySource propertyFileSource = (OriginTrackedMapPropertySource) loader
|
||||
.load("META-INF/spring.integration.properties", resource)
|
||||
.get(0);
|
||||
environment.getPropertySources().addLast(new IntegrationPropertiesPropertySource(propertyFileSource));
|
||||
}
|
||||
catch (IOException ex) {
|
||||
throw new IllegalStateException("Failed to load integration properties from " + resource, ex);
|
||||
}
|
||||
}
|
||||
|
||||
private static final class IntegrationPropertiesPropertySource extends PropertySource<Map<String, Object>>
|
||||
implements OriginLookup<String> {
|
||||
|
||||
private static final String PREFIX = "spring.integration.";
|
||||
|
||||
private static final Map<String, String> KEYS_MAPPING;
|
||||
|
||||
static {
|
||||
Map<String, String> mappings = new HashMap<>();
|
||||
mappings.put(PREFIX + "channel.auto-create", IntegrationProperties.CHANNELS_AUTOCREATE);
|
||||
mappings.put(PREFIX + "channel.max-unicast-subscribers",
|
||||
IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS);
|
||||
mappings.put(PREFIX + "channel.max-broadcast-subscribers",
|
||||
IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS);
|
||||
mappings.put(PREFIX + "error.require-subscribers", IntegrationProperties.ERROR_CHANNEL_REQUIRE_SUBSCRIBERS);
|
||||
mappings.put(PREFIX + "error.ignore-failures", IntegrationProperties.ERROR_CHANNEL_IGNORE_FAILURES);
|
||||
mappings.put(PREFIX + "endpoint.default-timeout", IntegrationProperties.ENDPOINTS_DEFAULT_TIMEOUT);
|
||||
mappings.put(PREFIX + "endpoint.throw-exception-on-late-reply",
|
||||
IntegrationProperties.THROW_EXCEPTION_ON_LATE_REPLY);
|
||||
mappings.put(PREFIX + "endpoint.read-only-headers", IntegrationProperties.READ_ONLY_HEADERS);
|
||||
mappings.put(PREFIX + "endpoint.no-auto-startup", IntegrationProperties.ENDPOINTS_NO_AUTO_STARTUP);
|
||||
KEYS_MAPPING = Collections.unmodifiableMap(mappings);
|
||||
}
|
||||
|
||||
private final OriginTrackedMapPropertySource delegate;
|
||||
|
||||
IntegrationPropertiesPropertySource(OriginTrackedMapPropertySource delegate) {
|
||||
super("META-INF/spring.integration.properties", delegate.getSource());
|
||||
this.delegate = delegate;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object getProperty(String name) {
|
||||
return this.delegate.getProperty(KEYS_MAPPING.get(name));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Origin getOrigin(String key) {
|
||||
return this.delegate.getOrigin(KEYS_MAPPING.get(key));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
/*
|
||||
* Copyright 2012-2025 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
|
||||
*
|
||||
* https://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.boot.integration.autoconfigure;
|
||||
|
||||
import org.springframework.integration.scheduling.PollerMetadata;
|
||||
|
||||
/**
|
||||
* Callback interface that can be implemented by beans wishing to customize the
|
||||
* {@link PollerMetadata} whilst retaining default auto-configuration.
|
||||
*
|
||||
* @author Yanming Zhou
|
||||
* @since 3.5.0
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface PollerMetadataCustomizer {
|
||||
|
||||
/**
|
||||
* Customize the {@link PollerMetadata}.
|
||||
* @param pollerMetadata the {@code PollerMetadata} to customize
|
||||
*/
|
||||
void customize(PollerMetadata pollerMetadata);
|
||||
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
/*
|
||||
* Copyright 2012-2019 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
|
||||
*
|
||||
* https://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.
|
||||
*/
|
||||
|
||||
/**
|
||||
* Auto-configuration for Spring Integration.
|
||||
*/
|
||||
package org.springframework.boot.integration.autoconfigure;
|
||||
@@ -0,0 +1,3 @@
|
||||
# Environment Post Processors
|
||||
org.springframework.boot.env.EnvironmentPostProcessor=\
|
||||
org.springframework.boot.integration.autoconfigure.IntegrationPropertiesEnvironmentPostProcessor
|
||||
@@ -0,0 +1 @@
|
||||
org.springframework.boot.integration.autoconfigure.IntegrationAutoConfiguration
|
||||
@@ -0,0 +1,697 @@
|
||||
/*
|
||||
* Copyright 2012-2025 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
|
||||
*
|
||||
* https://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.boot.integration.autoconfigure;
|
||||
|
||||
import java.beans.PropertyDescriptor;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import javax.management.MBeanServer;
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import io.micrometer.observation.ObservationRegistry;
|
||||
import io.rsocket.transport.ClientTransport;
|
||||
import io.rsocket.transport.netty.client.TcpClientTransport;
|
||||
import org.assertj.core.api.InstanceOfAssertFactories;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.condition.EnabledForJreRange;
|
||||
import org.junit.jupiter.api.condition.JRE;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.beans.DirectFieldAccessor;
|
||||
import org.springframework.beans.PropertyAccessorFactory;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
||||
import org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.task.TaskSchedulingAutoConfiguration;
|
||||
import org.springframework.boot.context.annotation.UserConfigurations;
|
||||
import org.springframework.boot.context.properties.source.MutuallyExclusiveConfigurationPropertiesException;
|
||||
import org.springframework.boot.flyway.autoconfigure.FlywayAutoConfiguration;
|
||||
import org.springframework.boot.integration.autoconfigure.IntegrationAutoConfiguration.IntegrationComponentScanConfiguration;
|
||||
import org.springframework.boot.jdbc.autoconfigure.DataSourceAutoConfiguration;
|
||||
import org.springframework.boot.jdbc.autoconfigure.DataSourceTransactionManagerAutoConfiguration;
|
||||
import org.springframework.boot.jdbc.autoconfigure.EmbeddedDataSourceConfiguration;
|
||||
import org.springframework.boot.jdbc.autoconfigure.JdbcTemplateAutoConfiguration;
|
||||
import org.springframework.boot.jdbc.init.DataSourceScriptDatabaseInitializer;
|
||||
import org.springframework.boot.rsocket.autoconfigure.RSocketMessagingAutoConfiguration;
|
||||
import org.springframework.boot.rsocket.autoconfigure.RSocketRequesterAutoConfiguration;
|
||||
import org.springframework.boot.rsocket.autoconfigure.RSocketServerAutoConfiguration;
|
||||
import org.springframework.boot.rsocket.autoconfigure.RSocketStrategiesAutoConfiguration;
|
||||
import org.springframework.boot.sql.init.DatabaseInitializationMode;
|
||||
import org.springframework.boot.sql.init.DatabaseInitializationSettings;
|
||||
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||
import org.springframework.boot.testsupport.assertj.SimpleAsyncTaskExecutorAssert;
|
||||
import org.springframework.boot.testsupport.classpath.resources.WithResource;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Primary;
|
||||
import org.springframework.core.task.SimpleAsyncTaskExecutor;
|
||||
import org.springframework.core.task.SyncTaskExecutor;
|
||||
import org.springframework.core.task.TaskExecutor;
|
||||
import org.springframework.integration.annotation.IntegrationComponentScan;
|
||||
import org.springframework.integration.annotation.MessagingGateway;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.QueueChannel;
|
||||
import org.springframework.integration.config.IntegrationManagementConfigurer;
|
||||
import org.springframework.integration.context.IntegrationContextUtils;
|
||||
import org.springframework.integration.core.MessageSource;
|
||||
import org.springframework.integration.endpoint.MessageProcessorMessageSource;
|
||||
import org.springframework.integration.gateway.RequestReplyExchanger;
|
||||
import org.springframework.integration.handler.BridgeHandler;
|
||||
import org.springframework.integration.handler.LoggingHandler;
|
||||
import org.springframework.integration.handler.MessageProcessor;
|
||||
import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport;
|
||||
import org.springframework.integration.monitor.IntegrationMBeanExporter;
|
||||
import org.springframework.integration.rsocket.ClientRSocketConnector;
|
||||
import org.springframework.integration.rsocket.IntegrationRSocketEndpoint;
|
||||
import org.springframework.integration.rsocket.ServerRSocketConnector;
|
||||
import org.springframework.integration.rsocket.ServerRSocketMessageHandler;
|
||||
import org.springframework.integration.scheduling.PollerMetadata;
|
||||
import org.springframework.integration.support.channel.HeaderChannelRegistry;
|
||||
import org.springframework.jdbc.BadSqlGrammarException;
|
||||
import org.springframework.jdbc.core.JdbcOperations;
|
||||
import org.springframework.jmx.export.MBeanExporter;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import org.springframework.scheduling.concurrent.SimpleAsyncTaskScheduler;
|
||||
import org.springframework.scheduling.support.CronTrigger;
|
||||
import org.springframework.scheduling.support.PeriodicTrigger;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
/**
|
||||
* Tests for {@link IntegrationAutoConfiguration}.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
* @author Stephane Nicoll
|
||||
* @author Vedran Pavic
|
||||
* @author Yong-Hyun Kim
|
||||
* @author Yanming Zhou
|
||||
*/
|
||||
class IntegrationAutoConfigurationTests {
|
||||
|
||||
private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
|
||||
.withConfiguration(AutoConfigurations.of(JmxAutoConfiguration.class, IntegrationAutoConfiguration.class));
|
||||
|
||||
@Test
|
||||
void integrationIsAvailable() {
|
||||
this.contextRunner.run((context) -> {
|
||||
assertThat(context).hasSingleBean(TestGateway.class);
|
||||
assertThat(context).hasSingleBean(IntegrationComponentScanConfiguration.class);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void explicitIntegrationComponentScan() {
|
||||
this.contextRunner.withUserConfiguration(CustomIntegrationComponentScanConfiguration.class).run((context) -> {
|
||||
assertThat(context).hasSingleBean(TestGateway.class);
|
||||
assertThat(context).doesNotHaveBean(IntegrationComponentScanConfiguration.class);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void noMBeanServerAvailable() {
|
||||
ApplicationContextRunner contextRunnerWithoutJmx = new ApplicationContextRunner()
|
||||
.withConfiguration(AutoConfigurations.of(IntegrationAutoConfiguration.class));
|
||||
contextRunnerWithoutJmx.run((context) -> {
|
||||
assertThat(context).hasSingleBean(TestGateway.class);
|
||||
assertThat(context).hasSingleBean(IntegrationComponentScanConfiguration.class);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void parentContext() {
|
||||
this.contextRunner.run((context) -> this.contextRunner.withParent(context)
|
||||
.withPropertyValues("spring.jmx.default_domain=org.foo")
|
||||
.run((child) -> assertThat(child).hasSingleBean(HeaderChannelRegistry.class)));
|
||||
}
|
||||
|
||||
@Test
|
||||
void enableJmxIntegration() {
|
||||
this.contextRunner.withPropertyValues("spring.jmx.enabled=true").run((context) -> {
|
||||
MBeanServer mBeanServer = context.getBean(MBeanServer.class);
|
||||
assertThat(mBeanServer.getDomains()).contains("org.springframework.integration",
|
||||
"org.springframework.boot.integration.autoconfigure");
|
||||
assertThat(context).hasBean(IntegrationManagementConfigurer.MANAGEMENT_CONFIGURER_NAME);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void jmxIntegrationIsDisabledByDefault() {
|
||||
this.contextRunner.run((context) -> {
|
||||
assertThat(context).doesNotHaveBean(MBeanServer.class);
|
||||
assertThat(context).hasSingleBean(IntegrationManagementConfigurer.class);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void customizeJmxDomain() {
|
||||
this.contextRunner.withPropertyValues("spring.jmx.enabled=true", "spring.jmx.default_domain=org.foo")
|
||||
.run((context) -> {
|
||||
MBeanServer mBeanServer = context.getBean(MBeanServer.class);
|
||||
assertThat(mBeanServer.getDomains()).contains("org.foo")
|
||||
.doesNotContain("org.springframework.integration", "org.springframework.integration.monitor");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void customJmxDomainUsingEnableIntegrationMBeanExport() {
|
||||
this.contextRunner.withConfiguration(UserConfigurations.of(CustomJmxDomainConfiguration.class))
|
||||
.withConfiguration(AutoConfigurations.of(JmxAutoConfiguration.class, IntegrationAutoConfiguration.class))
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(IntegrationMBeanExporter.class);
|
||||
IntegrationMBeanExporter exporter = context.getBean(IntegrationMBeanExporter.class);
|
||||
assertThat(exporter).hasFieldOrPropertyWithValue("domain", "foo.my");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void primaryExporterIsAllowed() {
|
||||
this.contextRunner.withPropertyValues("spring.jmx.enabled=true")
|
||||
.withUserConfiguration(CustomMBeanExporter.class)
|
||||
.run((context) -> {
|
||||
assertThat(context).getBeans(MBeanExporter.class).hasSize(2);
|
||||
assertThat(context.getBean(MBeanExporter.class)).isSameAs(context.getBean("myMBeanExporter"));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void integrationJdbcDataSourceInitializerEnabled() {
|
||||
this.contextRunner.withUserConfiguration(EmbeddedDataSourceConfiguration.class)
|
||||
.withConfiguration(AutoConfigurations.of(DataSourceTransactionManagerAutoConfiguration.class,
|
||||
JdbcTemplateAutoConfiguration.class, IntegrationAutoConfiguration.class))
|
||||
.withPropertyValues("spring.datasource.generate-unique-name=true",
|
||||
"spring.integration.jdbc.initialize-schema=always")
|
||||
.run((context) -> {
|
||||
IntegrationProperties properties = context.getBean(IntegrationProperties.class);
|
||||
assertThat(properties.getJdbc().getInitializeSchema()).isEqualTo(DatabaseInitializationMode.ALWAYS);
|
||||
JdbcOperations jdbc = context.getBean(JdbcOperations.class);
|
||||
assertThat(jdbc.queryForList("select * from INT_MESSAGE")).isEmpty();
|
||||
assertThat(jdbc.queryForList("select * from INT_GROUP_TO_MESSAGE")).isEmpty();
|
||||
assertThat(jdbc.queryForList("select * from INT_MESSAGE_GROUP")).isEmpty();
|
||||
assertThat(jdbc.queryForList("select * from INT_LOCK")).isEmpty();
|
||||
assertThat(jdbc.queryForList("select * from INT_CHANNEL_MESSAGE")).isEmpty();
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void whenIntegrationJdbcDataSourceInitializerIsEnabledThenFlywayCanBeUsed() {
|
||||
this.contextRunner.withUserConfiguration(EmbeddedDataSourceConfiguration.class)
|
||||
.withConfiguration(AutoConfigurations.of(DataSourceTransactionManagerAutoConfiguration.class,
|
||||
JdbcTemplateAutoConfiguration.class, IntegrationAutoConfiguration.class,
|
||||
FlywayAutoConfiguration.class))
|
||||
.withPropertyValues("spring.datasource.generate-unique-name=true",
|
||||
"spring.integration.jdbc.initialize-schema=always")
|
||||
.run((context) -> {
|
||||
IntegrationProperties properties = context.getBean(IntegrationProperties.class);
|
||||
assertThat(properties.getJdbc().getInitializeSchema()).isEqualTo(DatabaseInitializationMode.ALWAYS);
|
||||
JdbcOperations jdbc = context.getBean(JdbcOperations.class);
|
||||
assertThat(jdbc.queryForList("select * from INT_MESSAGE")).isEmpty();
|
||||
assertThat(jdbc.queryForList("select * from INT_GROUP_TO_MESSAGE")).isEmpty();
|
||||
assertThat(jdbc.queryForList("select * from INT_MESSAGE_GROUP")).isEmpty();
|
||||
assertThat(jdbc.queryForList("select * from INT_LOCK")).isEmpty();
|
||||
assertThat(jdbc.queryForList("select * from INT_CHANNEL_MESSAGE")).isEmpty();
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void integrationJdbcDataSourceInitializerDisabled() {
|
||||
this.contextRunner.withUserConfiguration(EmbeddedDataSourceConfiguration.class)
|
||||
.withConfiguration(AutoConfigurations.of(DataSourceTransactionManagerAutoConfiguration.class,
|
||||
JdbcTemplateAutoConfiguration.class, IntegrationAutoConfiguration.class))
|
||||
.withPropertyValues("spring.datasource.generate-unique-name=true",
|
||||
"spring.integration.jdbc.initialize-schema=never")
|
||||
.run((context) -> {
|
||||
assertThat(context).doesNotHaveBean(IntegrationDataSourceScriptDatabaseInitializer.class);
|
||||
IntegrationProperties properties = context.getBean(IntegrationProperties.class);
|
||||
assertThat(properties.getJdbc().getInitializeSchema()).isEqualTo(DatabaseInitializationMode.NEVER);
|
||||
JdbcOperations jdbc = context.getBean(JdbcOperations.class);
|
||||
assertThatExceptionOfType(BadSqlGrammarException.class)
|
||||
.isThrownBy(() -> jdbc.queryForList("select * from INT_MESSAGE"));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void integrationJdbcDataSourceInitializerEnabledByDefaultWithEmbeddedDb() {
|
||||
this.contextRunner.withUserConfiguration(EmbeddedDataSourceConfiguration.class)
|
||||
.withConfiguration(AutoConfigurations.of(DataSourceTransactionManagerAutoConfiguration.class,
|
||||
JdbcTemplateAutoConfiguration.class, IntegrationAutoConfiguration.class))
|
||||
.withPropertyValues("spring.datasource.generate-unique-name=true")
|
||||
.run((context) -> {
|
||||
IntegrationProperties properties = context.getBean(IntegrationProperties.class);
|
||||
assertThat(properties.getJdbc().getInitializeSchema()).isEqualTo(DatabaseInitializationMode.EMBEDDED);
|
||||
JdbcOperations jdbc = context.getBean(JdbcOperations.class);
|
||||
assertThat(jdbc.queryForList("select * from INT_MESSAGE")).isEmpty();
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void rsocketSupportEnabled() {
|
||||
this.contextRunner.withUserConfiguration(RSocketServerConfiguration.class)
|
||||
.withConfiguration(AutoConfigurations.of(RSocketServerAutoConfiguration.class,
|
||||
RSocketStrategiesAutoConfiguration.class, RSocketMessagingAutoConfiguration.class,
|
||||
RSocketRequesterAutoConfiguration.class, IntegrationAutoConfiguration.class))
|
||||
.withPropertyValues("spring.rsocket.server.port=0", "spring.integration.rsocket.client.port=0",
|
||||
"spring.integration.rsocket.client.host=localhost",
|
||||
"spring.integration.rsocket.server.message-mapping-enabled=true")
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(ClientRSocketConnector.class)
|
||||
.hasBean("clientRSocketConnector")
|
||||
.hasSingleBean(ServerRSocketConnector.class)
|
||||
.hasSingleBean(ServerRSocketMessageHandler.class)
|
||||
.hasSingleBean(RSocketMessageHandler.class);
|
||||
|
||||
ServerRSocketMessageHandler serverRSocketMessageHandler = context
|
||||
.getBean(ServerRSocketMessageHandler.class);
|
||||
assertThat(context).getBean(RSocketMessageHandler.class).isSameAs(serverRSocketMessageHandler);
|
||||
|
||||
ClientRSocketConnector clientRSocketConnector = context.getBean(ClientRSocketConnector.class);
|
||||
ClientTransport clientTransport = (ClientTransport) new DirectFieldAccessor(clientRSocketConnector)
|
||||
.getPropertyValue("clientTransport");
|
||||
|
||||
assertThat(clientTransport).isInstanceOf(TcpClientTransport.class);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void taskSchedulerIsNotOverridden() {
|
||||
this.contextRunner.withConfiguration(AutoConfigurations.of(TaskSchedulingAutoConfiguration.class))
|
||||
.withPropertyValues("spring.task.scheduling.thread-name-prefix=integration-scheduling-",
|
||||
"spring.task.scheduling.pool.size=3")
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(TaskScheduler.class);
|
||||
assertThat(context).getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)
|
||||
.hasFieldOrPropertyWithValue("threadNamePrefix", "integration-scheduling-")
|
||||
.hasFieldOrPropertyWithValue("scheduledExecutor.corePoolSize", 3);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void taskSchedulerCanBeCustomized() {
|
||||
TaskScheduler customTaskScheduler = mock(TaskScheduler.class);
|
||||
this.contextRunner.withConfiguration(AutoConfigurations.of(TaskSchedulingAutoConfiguration.class))
|
||||
.withBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class, () -> customTaskScheduler)
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(TaskScheduler.class);
|
||||
assertThat(context).getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)
|
||||
.isSameAs(customTaskScheduler);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void integrationGlobalPropertiesAutoConfigured() {
|
||||
String[] propertyValues = { "spring.integration.channel.auto-create=false",
|
||||
"spring.integration.channel.max-unicast-subscribers=2",
|
||||
"spring.integration.channel.max-broadcast-subscribers=3",
|
||||
"spring.integration.error.require-subscribers=false", "spring.integration.error.ignore-failures=false",
|
||||
"spring.integration.endpoint.defaultTimeout=60s",
|
||||
"spring.integration.endpoint.throw-exception-on-late-reply=true",
|
||||
"spring.integration.endpoint.read-only-headers=ignoredHeader",
|
||||
"spring.integration.endpoint.no-auto-startup=notStartedEndpoint,_org.springframework.integration.errorLogger" };
|
||||
assertThat(propertyValues).hasSameSizeAs(globalIntegrationPropertyNames());
|
||||
this.contextRunner.withPropertyValues(propertyValues).run((context) -> {
|
||||
assertThat(context).hasSingleBean(org.springframework.integration.context.IntegrationProperties.class);
|
||||
org.springframework.integration.context.IntegrationProperties integrationProperties = context
|
||||
.getBean(org.springframework.integration.context.IntegrationProperties.class);
|
||||
assertThat(integrationProperties.isChannelsAutoCreate()).isFalse();
|
||||
assertThat(integrationProperties.getChannelsMaxUnicastSubscribers()).isEqualTo(2);
|
||||
assertThat(integrationProperties.getChannelsMaxBroadcastSubscribers()).isEqualTo(3);
|
||||
assertThat(integrationProperties.isErrorChannelRequireSubscribers()).isFalse();
|
||||
assertThat(integrationProperties.isErrorChannelIgnoreFailures()).isFalse();
|
||||
assertThat(integrationProperties.getEndpointsDefaultTimeout()).isEqualTo(60000);
|
||||
assertThat(integrationProperties.isMessagingTemplateThrowExceptionOnLateReply()).isTrue();
|
||||
assertThat(integrationProperties.getReadOnlyHeaders()).containsOnly("ignoredHeader");
|
||||
assertThat(integrationProperties.getNoAutoStartupEndpoints()).containsOnly("notStartedEndpoint",
|
||||
"_org.springframework.integration.errorLogger");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void integrationGlobalPropertiesUseConsistentDefault() {
|
||||
List<PropertyAccessor> properties = List
|
||||
.of("isChannelsAutoCreate", "getChannelsMaxUnicastSubscribers", "getChannelsMaxBroadcastSubscribers",
|
||||
"isErrorChannelRequireSubscribers", "isErrorChannelIgnoreFailures", "getEndpointsDefaultTimeout",
|
||||
"isMessagingTemplateThrowExceptionOnLateReply", "getReadOnlyHeaders", "getNoAutoStartupEndpoints")
|
||||
.stream()
|
||||
.map(PropertyAccessor::new)
|
||||
.toList();
|
||||
assertThat(properties).hasSameSizeAs(globalIntegrationPropertyNames());
|
||||
org.springframework.integration.context.IntegrationProperties defaultIntegrationProperties = new org.springframework.integration.context.IntegrationProperties();
|
||||
this.contextRunner.run((context) -> {
|
||||
assertThat(context).hasSingleBean(org.springframework.integration.context.IntegrationProperties.class);
|
||||
org.springframework.integration.context.IntegrationProperties integrationProperties = context
|
||||
.getBean(org.springframework.integration.context.IntegrationProperties.class);
|
||||
properties.forEach((property) -> assertThat(property.get(integrationProperties))
|
||||
.isEqualTo(property.get(defaultIntegrationProperties)));
|
||||
});
|
||||
}
|
||||
|
||||
private List<String> globalIntegrationPropertyNames() {
|
||||
return Stream
|
||||
.of(PropertyAccessorFactory
|
||||
.forBeanPropertyAccess(new org.springframework.integration.context.IntegrationProperties())
|
||||
.getPropertyDescriptors())
|
||||
.map(PropertyDescriptor::getName)
|
||||
.filter((name) -> !"class".equals(name))
|
||||
.filter((name) -> !"taskSchedulerPoolSize".equals(name))
|
||||
.toList();
|
||||
}
|
||||
|
||||
@Test
|
||||
void integrationGlobalPropertiesUserBeanOverridesAutoConfiguration() {
|
||||
org.springframework.integration.context.IntegrationProperties userIntegrationProperties = new org.springframework.integration.context.IntegrationProperties();
|
||||
this.contextRunner.withPropertyValues()
|
||||
.withBean(IntegrationContextUtils.INTEGRATION_GLOBAL_PROPERTIES_BEAN_NAME,
|
||||
org.springframework.integration.context.IntegrationProperties.class,
|
||||
() -> userIntegrationProperties)
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(org.springframework.integration.context.IntegrationProperties.class);
|
||||
assertThat(context.getBean(org.springframework.integration.context.IntegrationProperties.class))
|
||||
.isSameAs(userIntegrationProperties);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
@WithResource(name = "META-INF/spring.integration.properties",
|
||||
content = "spring.integration.endpoints.noAutoStartup=testService*")
|
||||
void integrationGlobalPropertiesFromSpringIntegrationPropertiesFile() {
|
||||
this.contextRunner
|
||||
.withPropertyValues("spring.integration.channel.auto-create=false",
|
||||
"spring.integration.endpoint.read-only-headers=ignoredHeader")
|
||||
.withInitializer((applicationContext) -> new IntegrationPropertiesEnvironmentPostProcessor()
|
||||
.postProcessEnvironment(applicationContext.getEnvironment(), null))
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(org.springframework.integration.context.IntegrationProperties.class);
|
||||
org.springframework.integration.context.IntegrationProperties integrationProperties = context
|
||||
.getBean(org.springframework.integration.context.IntegrationProperties.class);
|
||||
assertThat(integrationProperties.isChannelsAutoCreate()).isFalse();
|
||||
assertThat(integrationProperties.getReadOnlyHeaders()).containsOnly("ignoredHeader");
|
||||
// See META-INF/spring.integration.properties
|
||||
assertThat(integrationProperties.getNoAutoStartupEndpoints()).containsOnly("testService*");
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void whenTheUserDefinesTheirOwnIntegrationDatabaseInitializerThenTheAutoConfiguredInitializerBacksOff() {
|
||||
this.contextRunner.withUserConfiguration(CustomIntegrationDatabaseInitializerConfiguration.class)
|
||||
.withConfiguration(AutoConfigurations.of(DataSourceAutoConfiguration.class,
|
||||
DataSourceTransactionManagerAutoConfiguration.class))
|
||||
.run((context) -> assertThat(context).hasSingleBean(IntegrationDataSourceScriptDatabaseInitializer.class)
|
||||
.doesNotHaveBean("integrationDataSourceScriptDatabaseInitializer")
|
||||
.hasBean("customInitializer"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void whenTheUserDefinesTheirOwnDatabaseInitializerThenTheAutoConfiguredIntegrationInitializerRemains() {
|
||||
this.contextRunner.withUserConfiguration(CustomDatabaseInitializerConfiguration.class)
|
||||
.withConfiguration(AutoConfigurations.of(DataSourceAutoConfiguration.class,
|
||||
DataSourceTransactionManagerAutoConfiguration.class))
|
||||
.run((context) -> assertThat(context).hasSingleBean(IntegrationDataSourceScriptDatabaseInitializer.class)
|
||||
.hasBean("customInitializer"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void defaultPoller() {
|
||||
this.contextRunner.withUserConfiguration(PollingConsumerConfiguration.class).run((context) -> {
|
||||
assertThat(context).hasSingleBean(PollerMetadata.class);
|
||||
PollerMetadata metadata = context.getBean(PollerMetadata.DEFAULT_POLLER, PollerMetadata.class);
|
||||
assertThat(metadata.getMaxMessagesPerPoll()).isEqualTo(PollerMetadata.MAX_MESSAGES_UNBOUNDED);
|
||||
assertThat(metadata.getReceiveTimeout()).isEqualTo(PollerMetadata.DEFAULT_RECEIVE_TIMEOUT);
|
||||
assertThat(metadata.getTrigger()).isNull();
|
||||
|
||||
GenericMessage<String> testMessage = new GenericMessage<>("test");
|
||||
context.getBean("testChannel", QueueChannel.class).send(testMessage);
|
||||
assertThat(context.getBean("sink", BlockingQueue.class).poll(10, TimeUnit.SECONDS)).isSameAs(testMessage);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void whenCustomPollerPropertiesAreSetThenTheyAreReflectedInPollerMetadata() {
|
||||
this.contextRunner.withUserConfiguration(PollingConsumerConfiguration.class)
|
||||
.withPropertyValues("spring.integration.poller.cron=* * * ? * *",
|
||||
"spring.integration.poller.max-messages-per-poll=1",
|
||||
"spring.integration.poller.receive-timeout=10s")
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(PollerMetadata.class);
|
||||
PollerMetadata metadata = context.getBean(PollerMetadata.DEFAULT_POLLER, PollerMetadata.class);
|
||||
assertThat(metadata.getMaxMessagesPerPoll()).isOne();
|
||||
assertThat(metadata.getReceiveTimeout()).isEqualTo(10000L);
|
||||
assertThat(metadata.getTrigger()).asInstanceOf(InstanceOfAssertFactories.type(CronTrigger.class))
|
||||
.satisfies((trigger) -> assertThat(trigger.getExpression()).isEqualTo("* * * ? * *"));
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void whenPollerPropertiesForMultipleTriggerTypesAreSetThenRefreshFails() {
|
||||
this.contextRunner
|
||||
.withPropertyValues("spring.integration.poller.cron=* * * ? * *",
|
||||
"spring.integration.poller.fixed-delay=1s")
|
||||
.run((context) -> assertThat(context).hasFailed()
|
||||
.getFailure()
|
||||
.hasRootCauseExactlyInstanceOf(MutuallyExclusiveConfigurationPropertiesException.class)
|
||||
.rootCause()
|
||||
.asInstanceOf(InstanceOfAssertFactories.type(MutuallyExclusiveConfigurationPropertiesException.class))
|
||||
.satisfies((ex) -> {
|
||||
assertThat(ex.getConfiguredNames()).containsExactlyInAnyOrder("spring.integration.poller.cron",
|
||||
"spring.integration.poller.fixed-delay");
|
||||
assertThat(ex.getMutuallyExclusiveNames()).containsExactlyInAnyOrder(
|
||||
"spring.integration.poller.cron", "spring.integration.poller.fixed-delay",
|
||||
"spring.integration.poller.fixed-rate");
|
||||
}));
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
void whenFixedDelayPollerPropertyIsSetThenItIsReflectedAsFixedDelayPropertyOfPeriodicTrigger() {
|
||||
this.contextRunner.withUserConfiguration(PollingConsumerConfiguration.class)
|
||||
.withPropertyValues("spring.integration.poller.fixed-delay=5000")
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(PollerMetadata.class);
|
||||
PollerMetadata metadata = context.getBean(PollerMetadata.DEFAULT_POLLER, PollerMetadata.class);
|
||||
assertThat(metadata.getTrigger()).asInstanceOf(InstanceOfAssertFactories.type(PeriodicTrigger.class))
|
||||
.satisfies((trigger) -> {
|
||||
assertThat(trigger.getPeriodDuration()).isEqualTo(Duration.ofSeconds(5));
|
||||
assertThat(trigger.isFixedRate()).isFalse();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void whenFixedRatePollerPropertyIsSetThenItIsReflectedAsFixedRatePropertyOfPeriodicTrigger() {
|
||||
this.contextRunner.withUserConfiguration(PollingConsumerConfiguration.class)
|
||||
.withPropertyValues("spring.integration.poller.fixed-rate=5000")
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(PollerMetadata.class);
|
||||
PollerMetadata metadata = context.getBean(PollerMetadata.DEFAULT_POLLER, PollerMetadata.class);
|
||||
assertThat(metadata.getTrigger()).asInstanceOf(InstanceOfAssertFactories.type(PeriodicTrigger.class))
|
||||
.satisfies((trigger) -> {
|
||||
assertThat(trigger.getPeriodDuration()).isEqualTo(Duration.ofSeconds(5));
|
||||
assertThat(trigger.isFixedRate()).isTrue();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
void integrationManagementLoggingIsEnabledByDefault() {
|
||||
this.contextRunner.withBean(DirectChannel.class, DirectChannel::new)
|
||||
.run((context) -> assertThat(context).getBean(DirectChannel.class)
|
||||
.extracting(DirectChannel::isLoggingEnabled)
|
||||
.isEqualTo(true));
|
||||
}
|
||||
|
||||
@Test
|
||||
void integrationManagementLoggingCanBeDisabled() {
|
||||
this.contextRunner.withPropertyValues("spring.integration.management.defaultLoggingEnabled=false")
|
||||
.withBean(DirectChannel.class, DirectChannel::new)
|
||||
.run((context) -> assertThat(context).getBean(DirectChannel.class)
|
||||
.extracting(DirectChannel::isLoggingEnabled)
|
||||
.isEqualTo(false));
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
void integrationManagementInstrumentedWithObservation() {
|
||||
this.contextRunner.withPropertyValues("spring.integration.management.observation-patterns=testHandler")
|
||||
.withBean("testHandler", LoggingHandler.class, () -> new LoggingHandler("warn"))
|
||||
.withBean(ObservationRegistry.class, ObservationRegistry::create)
|
||||
.withBean(BridgeHandler.class, BridgeHandler::new)
|
||||
.run((context) -> {
|
||||
assertThat(context).getBean("testHandler").extracting("observationRegistry").isNotNull();
|
||||
assertThat(context).getBean(BridgeHandler.class)
|
||||
.extracting("observationRegistry")
|
||||
.isEqualTo(ObservationRegistry.NOOP);
|
||||
});
|
||||
}
|
||||
|
||||
@Test
|
||||
@EnabledForJreRange(min = JRE.JAVA_21)
|
||||
void integrationVirtualThreadsEnabled() {
|
||||
this.contextRunner.withPropertyValues("spring.threads.virtual.enabled=true")
|
||||
.withConfiguration(AutoConfigurations.of(TaskSchedulingAutoConfiguration.class))
|
||||
.run((context) -> assertThat(context).hasSingleBean(TaskScheduler.class)
|
||||
.getBean(IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME, TaskScheduler.class)
|
||||
.isInstanceOf(SimpleAsyncTaskScheduler.class)
|
||||
.satisfies((taskScheduler) -> SimpleAsyncTaskExecutorAssert
|
||||
.assertThat((SimpleAsyncTaskExecutor) taskScheduler)
|
||||
.usesVirtualThreads()));
|
||||
}
|
||||
|
||||
@Test
|
||||
void pollerMetadataCanBeCustomizedViaPollerMetadataCustomizer() {
|
||||
TaskExecutor taskExecutor = new SyncTaskExecutor();
|
||||
this.contextRunner.withUserConfiguration(PollingConsumerConfiguration.class)
|
||||
.withBean(PollerMetadataCustomizer.class,
|
||||
() -> (pollerMetadata) -> pollerMetadata.setTaskExecutor(taskExecutor))
|
||||
.run((context) -> {
|
||||
assertThat(context).hasSingleBean(PollerMetadata.class);
|
||||
PollerMetadata metadata = context.getBean(PollerMetadata.DEFAULT_POLLER, PollerMetadata.class);
|
||||
assertThat(metadata.getTaskExecutor()).isSameAs(taskExecutor);
|
||||
});
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
static class CustomMBeanExporter {
|
||||
|
||||
@Bean
|
||||
@Primary
|
||||
MBeanExporter myMBeanExporter() {
|
||||
return mock(MBeanExporter.class);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@IntegrationComponentScan
|
||||
static class CustomIntegrationComponentScanConfiguration {
|
||||
|
||||
}
|
||||
|
||||
@MessagingGateway
|
||||
public interface TestGateway extends RequestReplyExchanger {
|
||||
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
static class MessageSourceConfiguration {
|
||||
|
||||
@Bean
|
||||
MessageSource<?> myMessageSource() {
|
||||
return new MessageProcessorMessageSource(mock(MessageProcessor.class));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
static class RSocketServerConfiguration {
|
||||
|
||||
@Bean
|
||||
IntegrationRSocketEndpoint mockIntegrationRSocketEndpoint() {
|
||||
return new IntegrationRSocketEndpoint() {
|
||||
|
||||
@Override
|
||||
public Mono<Void> handleMessage(Message<?> message) {
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String[] getPath() {
|
||||
return new String[] { "/rsocketTestPath" };
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
static class CustomIntegrationDatabaseInitializerConfiguration {
|
||||
|
||||
@Bean
|
||||
IntegrationDataSourceScriptDatabaseInitializer customInitializer(DataSource dataSource,
|
||||
IntegrationProperties properties) {
|
||||
return new IntegrationDataSourceScriptDatabaseInitializer(dataSource, properties.getJdbc());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
static class CustomDatabaseInitializerConfiguration {
|
||||
|
||||
@Bean
|
||||
DataSourceScriptDatabaseInitializer customInitializer(DataSource dataSource) {
|
||||
return new DataSourceScriptDatabaseInitializer(dataSource, new DatabaseInitializationSettings());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
static class PollingConsumerConfiguration {
|
||||
|
||||
@Bean
|
||||
QueueChannel testChannel() {
|
||||
return new QueueChannel();
|
||||
}
|
||||
|
||||
@Bean
|
||||
BlockingQueue<Message<?>> sink() {
|
||||
return new LinkedBlockingQueue<>();
|
||||
}
|
||||
|
||||
@ServiceActivator(inputChannel = "testChannel")
|
||||
@Bean
|
||||
MessageHandler handler(BlockingQueue<Message<?>> sink) {
|
||||
return sink::add;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
static class PropertyAccessor {
|
||||
|
||||
private final String name;
|
||||
|
||||
PropertyAccessor(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
Object get(org.springframework.integration.context.IntegrationProperties properties) {
|
||||
return ReflectionTestUtils.invokeMethod(properties, this.name);
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return this.name;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
@EnableIntegrationMBeanExport(defaultDomain = "foo.my")
|
||||
static class CustomJmxDomainConfiguration {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
/*
|
||||
* Copyright 2012-2023 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
|
||||
*
|
||||
* https://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.boot.integration.autoconfigure;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.boot.sql.init.DatabaseInitializationSettings;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.BDDMockito.then;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
/**
|
||||
* Tests for {@link IntegrationDataSourceScriptDatabaseInitializer}.
|
||||
*
|
||||
* @author Stephane Nicoll
|
||||
*/
|
||||
class IntegrationDataSourceScriptDatabaseInitializerTests {
|
||||
|
||||
@Test
|
||||
void getSettingsWithPlatformDoesNotTouchDataSource() {
|
||||
DataSource dataSource = mock(DataSource.class);
|
||||
IntegrationProperties properties = new IntegrationProperties();
|
||||
properties.getJdbc().setPlatform("test");
|
||||
DatabaseInitializationSettings settings = IntegrationDataSourceScriptDatabaseInitializer.getSettings(dataSource,
|
||||
properties.getJdbc());
|
||||
assertThat(settings.getSchemaLocations())
|
||||
.containsOnly("classpath:org/springframework/integration/jdbc/schema-test.sql");
|
||||
then(dataSource).shouldHaveNoInteractions();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
/*
|
||||
* Copyright 2012-2025 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
|
||||
*
|
||||
* https://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.boot.integration.autoconfigure;
|
||||
|
||||
import java.io.FileNotFoundException;
|
||||
import java.lang.reflect.Modifier;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.MethodSource;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.context.properties.bind.BindResult;
|
||||
import org.springframework.boot.context.properties.bind.Bindable;
|
||||
import org.springframework.boot.context.properties.bind.Binder;
|
||||
import org.springframework.boot.origin.Origin;
|
||||
import org.springframework.boot.origin.OriginLookup;
|
||||
import org.springframework.boot.origin.TextResourceOrigin;
|
||||
import org.springframework.boot.testsupport.classpath.resources.WithResource;
|
||||
import org.springframework.core.env.ConfigurableEnvironment;
|
||||
import org.springframework.core.env.MapPropertySource;
|
||||
import org.springframework.core.env.PropertySource;
|
||||
import org.springframework.core.env.StandardEnvironment;
|
||||
import org.springframework.core.io.ClassPathResource;
|
||||
import org.springframework.core.io.Resource;
|
||||
import org.springframework.integration.context.IntegrationProperties;
|
||||
import org.springframework.mock.env.MockEnvironment;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
import org.springframework.util.ClassUtils;
|
||||
import org.springframework.util.ReflectionUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
/**
|
||||
* Tests for {@link IntegrationPropertiesEnvironmentPostProcessor}.
|
||||
*
|
||||
* @author Stephane Nicoll
|
||||
*/
|
||||
class IntegrationPropertiesEnvironmentPostProcessorTests {
|
||||
|
||||
@Test
|
||||
@WithResource(name = "META-INF/spring.integration.properties",
|
||||
content = "spring.integration.endpoints.noAutoStartup=testService*")
|
||||
void postProcessEnvironmentAddPropertySource() {
|
||||
ConfigurableEnvironment environment = new StandardEnvironment();
|
||||
new IntegrationPropertiesEnvironmentPostProcessor().postProcessEnvironment(environment,
|
||||
mock(SpringApplication.class));
|
||||
assertThat(environment.getPropertySources().contains("META-INF/spring.integration.properties")).isTrue();
|
||||
assertThat(environment.getProperty("spring.integration.endpoint.no-auto-startup")).isEqualTo("testService*");
|
||||
}
|
||||
|
||||
@Test
|
||||
@WithResource(name = "META-INF/spring.integration.properties",
|
||||
content = "spring.integration.endpoints.noAutoStartup=testService*")
|
||||
void postProcessEnvironmentAddPropertySourceLast() {
|
||||
ConfigurableEnvironment environment = new StandardEnvironment();
|
||||
environment.getPropertySources()
|
||||
.addLast(new MapPropertySource("test",
|
||||
Collections.singletonMap("spring.integration.endpoint.no-auto-startup", "another*")));
|
||||
new IntegrationPropertiesEnvironmentPostProcessor().postProcessEnvironment(environment,
|
||||
mock(SpringApplication.class));
|
||||
assertThat(environment.getPropertySources().contains("META-INF/spring.integration.properties")).isTrue();
|
||||
assertThat(environment.getProperty("spring.integration.endpoint.no-auto-startup")).isEqualTo("another*");
|
||||
}
|
||||
|
||||
@Test
|
||||
void registerIntegrationPropertiesPropertySourceWithUnknownResourceThrowsException() {
|
||||
ConfigurableEnvironment environment = new StandardEnvironment();
|
||||
ClassPathResource unknown = new ClassPathResource("does-not-exist.properties", getClass());
|
||||
assertThatIllegalStateException()
|
||||
.isThrownBy(() -> new IntegrationPropertiesEnvironmentPostProcessor()
|
||||
.registerIntegrationPropertiesPropertySource(environment, unknown))
|
||||
.withCauseInstanceOf(FileNotFoundException.class)
|
||||
.withMessageContaining(unknown.toString());
|
||||
}
|
||||
|
||||
@Test
|
||||
void registerIntegrationPropertiesPropertySourceWithResourceAddPropertySource() {
|
||||
ConfigurableEnvironment environment = new StandardEnvironment();
|
||||
new IntegrationPropertiesEnvironmentPostProcessor().registerIntegrationPropertiesPropertySource(environment,
|
||||
new ClassPathResource("spring.integration.properties", getClass()));
|
||||
assertThat(environment.getProperty("spring.integration.channel.auto-create", Boolean.class)).isFalse();
|
||||
assertThat(environment.getProperty("spring.integration.channel.max-unicast-subscribers", Integer.class))
|
||||
.isEqualTo(4);
|
||||
assertThat(environment.getProperty("spring.integration.channel.max-broadcast-subscribers", Integer.class))
|
||||
.isEqualTo(6);
|
||||
assertThat(environment.getProperty("spring.integration.error.require-subscribers", Boolean.class)).isFalse();
|
||||
assertThat(environment.getProperty("spring.integration.error.ignore-failures", Boolean.class)).isFalse();
|
||||
assertThat(environment.getProperty("spring.integration.endpoint.throw-exception-on-late-reply", Boolean.class))
|
||||
.isTrue();
|
||||
assertThat(environment.getProperty("spring.integration.endpoint.read-only-headers", String.class))
|
||||
.isEqualTo("header1,header2");
|
||||
assertThat(environment.getProperty("spring.integration.endpoint.no-auto-startup", String.class))
|
||||
.isEqualTo("testService,anotherService");
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void registerIntegrationPropertiesPropertySourceWithResourceCanRetrieveOrigin() {
|
||||
ConfigurableEnvironment environment = new StandardEnvironment();
|
||||
ClassPathResource resource = new ClassPathResource("spring.integration.properties", getClass());
|
||||
new IntegrationPropertiesEnvironmentPostProcessor().registerIntegrationPropertiesPropertySource(environment,
|
||||
resource);
|
||||
PropertySource<?> ps = environment.getPropertySources().get("META-INF/spring.integration.properties");
|
||||
assertThat(ps).isInstanceOf(OriginLookup.class);
|
||||
OriginLookup<String> originLookup = (OriginLookup<String>) ps;
|
||||
assertThat(originLookup.getOrigin("spring.integration.channel.auto-create"))
|
||||
.satisfies(textOrigin(resource, 0, 39));
|
||||
assertThat(originLookup.getOrigin("spring.integration.channel.max-unicast-subscribers"))
|
||||
.satisfies(textOrigin(resource, 1, 50));
|
||||
assertThat(originLookup.getOrigin("spring.integration.channel.max-broadcast-subscribers"))
|
||||
.satisfies(textOrigin(resource, 2, 52));
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("unchecked")
|
||||
void hasMappingsForAllMappableProperties() throws Exception {
|
||||
Class<?> propertySource = ClassUtils.forName("%s.IntegrationPropertiesPropertySource"
|
||||
.formatted(IntegrationPropertiesEnvironmentPostProcessor.class.getName()), getClass().getClassLoader());
|
||||
Map<String, String> mappings = (Map<String, String>) ReflectionTestUtils.getField(propertySource,
|
||||
"KEYS_MAPPING");
|
||||
assertThat(mappings.values()).containsExactlyInAnyOrderElementsOf(integrationPropertyNames());
|
||||
}
|
||||
|
||||
private static List<String> integrationPropertyNames() {
|
||||
List<String> propertiesToMap = new ArrayList<>();
|
||||
ReflectionUtils.doWithFields(IntegrationProperties.class, (field) -> {
|
||||
String value = (String) ReflectionUtils.getField(field, null);
|
||||
if (value.startsWith(IntegrationProperties.INTEGRATION_PROPERTIES_PREFIX)
|
||||
&& value.length() > IntegrationProperties.INTEGRATION_PROPERTIES_PREFIX.length()) {
|
||||
propertiesToMap.add(value);
|
||||
}
|
||||
}, (field) -> Modifier.isStatic(field.getModifiers()) && field.getType().equals(String.class));
|
||||
propertiesToMap.remove(IntegrationProperties.TASK_SCHEDULER_POOL_SIZE);
|
||||
return propertiesToMap;
|
||||
}
|
||||
|
||||
@MethodSource("mappedConfigurationProperties")
|
||||
@ParameterizedTest
|
||||
void mappedPropertiesExistOnBootsIntegrationProperties(String mapping) {
|
||||
Bindable<org.springframework.boot.integration.autoconfigure.IntegrationProperties> bindable = Bindable
|
||||
.of(org.springframework.boot.integration.autoconfigure.IntegrationProperties.class);
|
||||
MockEnvironment environment = new MockEnvironment().withProperty(mapping,
|
||||
(mapping.contains("max") || mapping.contains("timeout")) ? "1" : "true");
|
||||
BindResult<org.springframework.boot.integration.autoconfigure.IntegrationProperties> result = Binder
|
||||
.get(environment)
|
||||
.bind("spring.integration", bindable);
|
||||
assertThat(result.isBound()).isTrue();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static Collection<String> mappedConfigurationProperties() {
|
||||
try {
|
||||
Class<?> propertySource = ClassUtils.forName("%s.IntegrationPropertiesPropertySource"
|
||||
.formatted(IntegrationPropertiesEnvironmentPostProcessor.class.getName()), null);
|
||||
Map<String, String> mappings = (Map<String, String>) ReflectionTestUtils.getField(propertySource,
|
||||
"KEYS_MAPPING");
|
||||
return mappings.keySet();
|
||||
}
|
||||
catch (Exception ex) {
|
||||
throw new RuntimeException(ex);
|
||||
}
|
||||
}
|
||||
|
||||
private Consumer<Origin> textOrigin(Resource resource, int line, int column) {
|
||||
return (origin) -> {
|
||||
assertThat(origin).isInstanceOf(TextResourceOrigin.class);
|
||||
TextResourceOrigin textOrigin = (TextResourceOrigin) origin;
|
||||
assertThat(textOrigin.getResource()).isEqualTo(resource);
|
||||
assertThat(textOrigin.getLocation().getLine()).isEqualTo(line);
|
||||
assertThat(textOrigin.getLocation().getColumn()).isEqualTo(column);
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
spring.integration.channels.autoCreate=false
|
||||
spring.integration.channels.maxUnicastSubscribers=4
|
||||
spring.integration.channels.maxBroadcastSubscribers=6
|
||||
spring.integration.channels.error.requireSubscribers=false
|
||||
spring.integration.channels.error.ignoreFailures=false
|
||||
spring.integration.messagingTemplate.throwExceptionOnLateReply=true
|
||||
spring.integration.readOnly.headers=header1,header2
|
||||
spring.integration.endpoints.noAutoStartup=testService,anotherService
|
||||
Reference in New Issue
Block a user