From 54b0800d08613aa844fb9997397949e2ebc5e933 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 27 Jun 2017 14:40:27 -0400 Subject: [PATCH] INT-4280: Add `taskScheduler()` to Java DSL JIRA: https://jira.spring.io/browse/INT-4280 To let end-user easily to inject custom `TaskScheduler` to the endpoint add `ConsumerEndpointSpec.taskScheduler()` option * Address PR comments --- .../config/ConsumerEndpointFactoryBean.java | 10 +++++ .../integration/dsl/ConsumerEndpointSpec.java | 17 ++++++++ .../dsl/CorrelationHandlerSpec.java | 4 +- .../dsl/flows/IntegrationFlowTests.java | 41 +++++++++++++++++++ 4 files changed, 71 insertions(+), 1 deletion(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 2a31913ad6..a7fdbff18a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -49,6 +49,7 @@ import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; 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.CollectionUtils; import org.springframework.util.StringUtils; @@ -99,6 +100,8 @@ public class ConsumerEndpointFactoryBean private volatile DestinationResolver channelResolver; + private TaskScheduler taskScheduler; + public void setHandler(MessageHandler handler) { Assert.notNull(handler, "handler must not be null"); synchronized (this.handlerMonitor) { @@ -160,6 +163,10 @@ public class ConsumerEndpointFactoryBean this.adviceChain = adviceChain; } + public void setTaskScheduler(TaskScheduler taskScheduler) { + this.taskScheduler = taskScheduler; + } + @Override public void afterPropertiesSet() throws Exception { if (this.beanName == null) { @@ -296,6 +303,9 @@ public class ConsumerEndpointFactoryBean phase = Integer.MAX_VALUE / 2; } this.endpoint.setPhase(phase); + if (this.taskScheduler != null) { + this.endpoint.setTaskScheduler(this.taskScheduler); + } this.endpoint.afterPropertiesSet(); this.initialized = true; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java index 15a06a170d..765b72a1e2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java @@ -30,9 +30,11 @@ import org.springframework.integration.router.AbstractMessageRouter; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.transaction.TransactionInterceptorBuilder; import org.springframework.messaging.MessageHandler; +import org.springframework.scheduling.TaskScheduler; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.util.Assert; import reactor.util.function.Tuple2; @@ -78,6 +80,21 @@ public abstract class ConsumerEndpointSpec, return _this(); } + /** + * Configure a {@link TaskScheduler} for scheduling tasks, for example in the + * 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. + * @return the endpoint spec. + * @see org.springframework.integration.context.IntegrationContextUtils#getTaskScheduler + */ + public S taskScheduler(TaskScheduler taskScheduler) { + Assert.notNull(taskScheduler, "'taskScheduler' must not be null"); + this.endpointFactoryBean.setTaskScheduler(taskScheduler); + return _this(); + } + /** * Configure a list of {@link Advice} objects to be applied, in nested order, to the * endpoint's handler. The advice objects are applied only to the handler. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java index 6a70f6cac1..a050abfe68 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java @@ -134,8 +134,10 @@ public abstract class CorrelationHandlerSpec threadNameReference = new AtomicReference<>(); + CountDownLatch resultLatch = new CountDownLatch(1); + this.dedicatedResults.subscribe(m -> { + threadNameReference.set(Thread.currentThread().getName()); + resultLatch.countDown(); + }); + + this.dedicatedQueueChannel.send(new GenericMessage<>("foo")); + + assertTrue(resultLatch.await(10, TimeUnit.SECONDS)); + + assertEquals("dedicatedTaskScheduler-1", threadNameReference.get()); + } + @MessagingGateway public interface ControlBusGateway { @@ -721,6 +746,22 @@ public class IntegrationFlowTests { } + @Bean + public IntegrationFlow dedicatedPollingThreadFlow() { + return IntegrationFlows.from(MessageChannels.queue("dedicatedQueueChannel")) + .bridge(e -> e + .poller(Pollers.fixedDelay(0).receiveTimeout(-1)) + .taskScheduler(dedicatedTaskScheduler())) + .channel("dedicatedResults") + .get(); + } + + + @Bean + public TaskScheduler dedicatedTaskScheduler() { + return new ThreadPoolTaskScheduler(); + } + } @MessagingGateway