INT-4330: Add support for Supplier MessageSource

JIRA: https://jira.spring.io/browse/INT-4330

- updated documentation
- updated test

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java

@@ -19,6 +19,7 @@ package org.springframework.integration.config.annotation;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
 import java.util.List;
+import java.util.function.Supplier;
 
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
@@ -33,12 +34,14 @@ import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.util.MessagingAnnotationUtils;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * Post-processor for Methods annotated with {@link InboundChannelAdapter @InboundChannelAdapter}.
  *
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Oleg Zhurakousky
  * @since 4.0
  */
 public class InboundChannelAdapterAnnotationPostProcessor extends
@@ -80,20 +83,28 @@ public class InboundChannelAdapterAnnotationPostProcessor extends
 	}
 
 	private MessageSource<?> createMessageSource(Object bean, String beanName, Method method) {
+		MessageSource<?> messageSource = null;
 		if (AnnotatedElementUtils.isAnnotated(method, Bean.class.getName())) {
 			Object target = this.resolveTargetBeanFromMethodWithBeanAnnotation(method);
-			Assert.isInstanceOf(MessageSource.class, target, "The '" + this.annotationType + "' on @Bean method " +
-					"level is allowed only for: " + MessageSource.class.getName() + "beans");
-			return (MessageSource<?>) target;
+			Assert.isTrue(target instanceof MessageSource || target instanceof Supplier, "The '" + this.annotationType + "' on @Bean method " +
+					"level is allowed only for: " + MessageSource.class.getName() + " or " + Supplier.class.getName() + " beans");
+			if (target instanceof MessageSource<?>) {
+				messageSource = (MessageSource<?>) target;
+			}
+			else {
+				method = ReflectionUtils.findMethod(Supplier.class, "get");
+				bean = target;
+			}
 		}
-		else {
-			MethodInvokingMessageSource messageSource = new MethodInvokingMessageSource();
-			messageSource.setObject(bean);
-			messageSource.setMethod(method);
+		if (messageSource == null) {
+			MethodInvokingMessageSource methodInvokingMessageSource = new MethodInvokingMessageSource();
+			methodInvokingMessageSource.setObject(bean);
+			methodInvokingMessageSource.setMethod(method);
 			String messageSourceBeanName = this.generateHandlerBeanName(beanName, method);
-			this.beanFactory.registerSingleton(messageSourceBeanName, messageSource);
-			return  (MessageSource<?>) this.beanFactory.initializeBean(messageSource, messageSourceBeanName);
+			this.beanFactory.registerSingleton(messageSourceBeanName, methodInvokingMessageSource);
+			messageSource =  (MessageSource<?>) this.beanFactory.initializeBean(methodInvokingMessageSource, messageSourceBeanName);
 		}
+		return messageSource;
 	}
 
 	@Override

spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationsWithBeanAnnotationTests.java

@@ -28,13 +28,14 @@ import static org.junit.Assert.fail;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
 
 import javax.annotation.Resource;
 
 import org.hamcrest.Matchers;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -84,6 +85,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Oleg Zhurakousky
  * @since 4.0
  */
 @ContextConfiguration(classes = MessagingAnnotationsWithBeanAnnotationTests.ContextConfiguration.class)
@@ -92,7 +94,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 public class MessagingAnnotationsWithBeanAnnotationTests {
 
 	@Autowired
-	private SourcePollingChannelAdapter sourcePollingChannelAdapter;
+	private SourcePollingChannelAdapter[] sourcePollingChannelAdapters;
 
 	@Autowired
 	private PollableChannel discardChannel;
@@ -125,7 +127,8 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
 
 	@Test
 	public void testMessagingAnnotationsFlow() {
-		this.sourcePollingChannelAdapter.start();
+		Stream.of(this.sourcePollingChannelAdapters).forEach(a -> a.start());
+		//this.sourcePollingChannelAdapter.start();
 		for (int i = 0; i < 10; i++) {
 			Message<?> receive = this.discardChannel.receive(10000);
 			assertNotNull(receive);
@@ -194,6 +197,13 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
 			return () -> new GenericMessage<>(counter.incrementAndGet());
 		}
 
+		@Bean
+		@InboundChannelAdapter(value = "routerChannel", autoStartup = "false",
+				poller = @Poller(fixedRate = "10", maxMessagesPerPoll = "1", errorChannel = "counterErrorChannel"))
+		public Supplier<Integer> counterMessageSupplier(final AtomicInteger counter) {
+			return () -> counter.incrementAndGet();
+		}
+
 		@Bean
 		public PollableChannel counterErrorChannel() {
 			return new QueueChannel();

src/reference/asciidoc/configuration.adoc

@@ -569,6 +569,26 @@ public class MyFlowConfiguration {
 }
 ----
 
+Starting with _version 5.0_, support is also provided for `@Bean` annotated `InboundChannelAdapter`s that return `java.util.function.Supplier` which can produce either a POJO or a `Message`:
+[source,java]
+----
+@Configuration
+@EnableIntegration
+public class MyFlowConfiguration {
+
+    @Bean
+    @InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
+    public Supplier<String> pojoSupplier() {
+        return () -> "foo";
+    }
+	
+    @Bean
+    @InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "1000"))
+    public Supplier<Message<String>> messageSupplier() {
+        return () -> new GenericMessage<>("foo");
+    }
+----
+
 The meta-annotation rules work on `@Bean` methods as well (`@MyServiceActivator` above can be applied to a `@Bean` definition).
 
 NOTE: When using these annotations on consumer `@Bean` definitions, if the bean definition returns an appropriate