From 318741d1fd2bb185e67f02b8ffe430c5414a9fea Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Sat, 19 Aug 2017 09:39:03 -0400 Subject: [PATCH] INT-4332 updated DSL with Supplier support - add support for Supplier as sources of messages to the DSL - add tests and javadocs --- .../integration/dsl/IntegrationFlows.java | 27 ++++++++ .../dsl/flows/IntegrationFlowTests.java | 67 +++++++++++++++++++ 2 files changed, 94 insertions(+) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 7874fff9f1..bde4c5eade 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -17,6 +17,7 @@ package org.springframework.integration.dsl; import java.util.function.Consumer; +import java.util.function.Supplier; import org.reactivestreams.Publisher; @@ -28,6 +29,7 @@ import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototy import org.springframework.integration.dsl.support.MessageChannelReference; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.endpoint.MethodInvokingMessageSource; +import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.gateway.AnnotationGatewayProxyFactoryBean; import org.springframework.integration.gateway.GatewayProxyFactoryBean; import org.springframework.integration.gateway.MessagingGatewaySupport; @@ -40,6 +42,7 @@ import org.springframework.util.Assert; * * @author Artem Bilan * @author Gary Russell + * @author Oleg Zhurakousky * * @since 5.0 * @@ -146,6 +149,30 @@ public final class IntegrationFlows { return from(service, methodName, null); } + /** + * Provides {@link Supplier} as source of messages to the integration flow which will + * be triggered by the application context's default poller (which must be declared). + * @param messageSource the {@link Supplier} to populate. + * @return new {@link IntegrationFlowBuilder}. + * @see Supplier + */ + public static IntegrationFlowBuilder from(Supplier messageSource) { + return from(messageSource, "get", null); + } + + /** + * Provides {@link Supplier} as source of messages to the integration flow. + * which will be triggered by a provided {@link SourcePollingChannelAdapter}. + * @param messageSource the {@link Supplier} to populate. + * @param endpointConfigurer the {@link Consumer} to provide more options for the + * {@link org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean}. + * @return new {@link IntegrationFlowBuilder}. + * @see Supplier + */ + public static IntegrationFlowBuilder from(Supplier messageSource, Consumer endpointConfigurer) { + return from(messageSource, "get", endpointConfigurer); + } + /** * Populate the provided {@link MethodInvokingMessageSource} for the method of the provided service. * The {@link org.springframework.integration.dsl.IntegrationFlow} {@code startMessageSource}. 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 880ec9ea10..b9a88b1539 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 @@ -95,6 +95,7 @@ import org.springframework.test.context.junit4.SpringRunner; * @author Artem Bilan * @author Tim Ysewyn * @author Gary Russell + * @author Oleg Zhurakousky * * @since 5.0 */ @@ -124,6 +125,14 @@ public class IntegrationFlowTests { @Qualifier("successChannel") private PollableChannel successChannel; + @Autowired + @Qualifier("suppliedChannel") + private PollableChannel suppliedChannel; + + @Autowired + @Qualifier("suppliedChannel2") + private PollableChannel suppliedChannel2; + @Autowired @Qualifier("bridgeFlowInput") private PollableChannel bridgeFlowInput; @@ -167,6 +176,17 @@ public class IntegrationFlowTests { @Qualifier("gatewayError") private PollableChannel gatewayError; + @Test + public void testWithSupplierMessageSourceImpliedPoller() { + assertEquals("FOO", this.suppliedChannel.receive(1000).getPayload()); + } + + @Test + public void testWithSupplierMessageSourceProvidedPoller() { + assertNull(this.suppliedChannel2.receive(100)); + assertEquals("FOO", this.suppliedChannel2.receive(2000).getPayload()); + } + @Test public void testDirectFlow() { assertTrue(this.beanFactory.containsBean("filter")); @@ -459,6 +479,53 @@ public class IntegrationFlowTests { void send(String command); } + @Configuration + @EnableIntegration + public static class SupplierContextConfiguration1 { + @Bean + public IntegrationFlow supplierFlow() { + return IntegrationFlows.from(() -> "foo") + .transform(p -> p.toUpperCase()) + .channel("suppliedChannel") + .get(); + } + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerMetadata poller() { + return Pollers.fixedRate(100).get(); + } + + @Bean(name = IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME) + public TaskScheduler taskScheduler() { + ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler(); + threadPoolTaskScheduler.setPoolSize(100); + return threadPoolTaskScheduler; + } + + + @Bean + public MessageChannel suppliedChannel() { + return MessageChannels.queue(10).get(); + } + } + + @Configuration + @EnableIntegration + public static class SupplierContextConfiguration2 { + @Bean + public IntegrationFlow supplierFlow2() { + return IntegrationFlows.from(() -> "foo", c -> c.poller(Pollers.fixedDelay(400, 300).maxMessagesPerPoll(1))) + .transform(p -> p.toUpperCase()) + .channel("suppliedChannel2") + .get(); + } + + @Bean + public MessageChannel suppliedChannel2() { + return MessageChannels.queue(10).get(); + } + } + @Configuration @EnableIntegration public static class ContextConfiguration {