INT-4332 updated DSL with Supplier support

- add support for Supplier as sources of messages to the DSL
- add tests and javadocs

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 <T> IntegrationFlowBuilder from(Supplier<T> messageSource) {
+		return from(messageSource, "get", null);
+	}
+
+	/**
+	 * Provides {@link Supplier} as source of messages to the integration flow.
+	 * which will be triggered by a <b>provided</b> {@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 <T> IntegrationFlowBuilder from(Supplier<T> messageSource, Consumer<SourcePollingChannelAdapterSpec> 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}.

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")
+					.<String, String>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)))
+					.<String, String>transform(p -> p.toUpperCase())
+					.channel("suppliedChannel2")
+					.get();
+		}
+
+		@Bean
+		public MessageChannel suppliedChannel2() {
+			return MessageChannels.queue(10).get();
+		}
+	}
+
 	@Configuration
 	@EnableIntegration
 	public static class ContextConfiguration {