From d5ab03c0c770ace8ce7e12234e221d3a997c3021 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 14 Oct 2024 17:22:29 -0400 Subject: [PATCH] GH-9524: Expose `SourcePollingChannelAdapterSpec.taskScheduler` Fixes: #9524 Issue link: https://github.com/spring-projects/spring-integration/issues/9524 It is useful in some use-cases to be able to inject a custom `TaskScheduler` (e.g. with a `TaskDecorator`) into a source polling channel adapter. * Add `SourcePollingChannelAdapterFactoryBean.setTaskScheduler()` and call it from the `SourcePollingChannelAdapterSpec.taskScheduler()` * Fix JavaDocs typos in the `ConsumerEndpointSpec` * Test custom `TaskScheduler` usage and mention new option in the `whats-new.adoc` --- .../SourcePollingChannelAdapterFactoryBean.java | 15 +++++++++++++++ .../integration/dsl/ConsumerEndpointSpec.java | 6 +++--- .../dsl/SourcePollingChannelAdapterSpec.java | 14 +++++++++++++- .../dsl/flows/IntegrationFlowTests.java | 17 +++++++++++++++-- .../antora/modules/ROOT/pages/whats-new.adoc | 2 ++ 5 files changed, 48 insertions(+), 6 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java index ece0dd82b5..b34aa478d6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java @@ -34,6 +34,7 @@ import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver; import org.springframework.messaging.core.DestinationResolver; +import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -75,6 +76,8 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean, /** * Configure a {@link TaskScheduler} for scheduling tasks, for example in the - * Polling Consumer. By default the global {@code ThreadPoolTaskScheduler} bean is used. + * Polling Consumer. By default, the global {@code ThreadPoolTaskScheduler} bean is used. * This configuration is useful when there are requirements to dedicate particular threads * for polling task, for example. * @param taskScheduler the {@link TaskScheduler} to use. @@ -144,7 +144,7 @@ public abstract class ConsumerEndpointSpec, /** * Configure a list of {@link MethodInterceptor} objects to be applied, in nested order, to the * endpoint's handler. The advice objects are applied to the {@code handleMessage()} method - * and therefore to the whole sub-flow afterwards. + * and therefore to the whole sub-flow afterward. * @param interceptors the advice chain. * @return the endpoint spec. * @since 5.3 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java index 594cf4641f..c739692416 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-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. @@ -20,6 +20,7 @@ import org.springframework.integration.config.SourcePollingChannelAdapterFactory import org.springframework.integration.core.MessageSource; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.lang.Nullable; +import org.springframework.scheduling.TaskScheduler; /** * @author Artem Bilan @@ -60,4 +61,15 @@ public class SourcePollingChannelAdapterSpec extends return this; } + /** + * Set a {@link TaskScheduler} for polling tasks. + * @param taskScheduler the {@link TaskScheduler} for polling tasks. + * @return the spec + * @since 6.4 + */ + public SourcePollingChannelAdapterSpec taskScheduler(TaskScheduler taskScheduler) { + this.endpointFactoryBean.setTaskScheduler(taskScheduler); + return this; + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java index 0450468a65..6224155d25 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java @@ -81,6 +81,7 @@ import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.MutableMessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.transformer.PayloadSerializingTransformer; import org.springframework.integration.util.NoBeansOverrideAnnotationConfigContextLoader; import org.springframework.messaging.Message; @@ -94,6 +95,7 @@ import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; +import org.springframework.scheduling.concurrent.SimpleAsyncTaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.stereotype.Component; @@ -187,10 +189,15 @@ public class IntegrationFlowTests { @Autowired AbstractEndpoint stringSupplierEndpoint; + @Autowired + TaskScheduler customScheduler; + @Test public void testWithSupplierMessageSourceImpliedPoller() { assertThat(this.stringSupplierEndpoint.isAutoStartup()).isFalse(); assertThat(this.stringSupplierEndpoint.isRunning()).isFalse(); + assertThat(TestUtils.getPropertyValue(this.stringSupplierEndpoint, "taskScheduler")) + .isSameAs(this.customScheduler); this.stringSupplierEndpoint.start(); assertThat(this.suppliedChannel.receive(10000).getPayload()).isEqualTo("FOO"); } @@ -569,8 +576,14 @@ public class IntegrationFlowTests { } @Bean - public IntegrationFlow supplierFlow() { - return IntegrationFlow.fromSupplier(stringSupplier(), c -> c.id("stringSupplierEndpoint")) + public TaskScheduler customScheduler() { + return new SimpleAsyncTaskScheduler(); + } + + @Bean + public IntegrationFlow supplierFlow(TaskScheduler customScheduler) { + return IntegrationFlow.fromSupplier(stringSupplier(), + c -> c.id("stringSupplierEndpoint").taskScheduler(customScheduler)) .transform(toUpperCaseFunction()) .channel("suppliedChannel") .get(); diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 4187b3b7b9..3a0b552d4c 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -30,6 +30,8 @@ See xref:spel.adoc[SpEL Support] for more information. [[x6.4-general]] === General Changes +The Java DSL `SourcePollingChannelAdapterSpec` can now be configured with a custom `TaskScheduler` + [[x6.4-remote-files-changes]] === Remote File Adapters Changes