GH-1956 Add support for PollableSource in functional programing model

This feature effectively provides identical path to create PollableSource as it was in annotation-based programming model.
However, unlike creating and  providing binding intreface all that is required from user is 'spring.cloud.stream.pollable-source' property

Resolves #1956

docs/src/main/asciidoc/spring-cloud-stream.adoc

@@ -1242,21 +1242,18 @@ support.
 ===== Overview
 
 When using polled consumers, you poll the `PollableMessageSource` on demand.
-Consider the following example of a polled consumer:
+To define binding for polled consumer you need to provide `spring.cloud.stream.pollable-source` property.
 
-[source,java]
+Consider the following example of a polled consumer binding:
+
+[source,text]
 ----
-public interface PolledConsumer {
-
-    @Input
-    PollableMessageSource destIn();
-
-    @Output
-    MessageChannel destOut();
-
-}
+--spring.cloud.stream.pollable-source=myDestination
 ----
 
+The pollable-source name `myDestination` in the preceding example will result in `myDestination-in-0` binding name to stay 
+consistent with functional programming model.
+
 Given the polled consumer in the preceding example, you might use it as follows:
 
 [source,java]
@@ -1348,6 +1345,7 @@ boolean result = pollableSource.poll(received -> {
 		}, new ParameterizedTypeReference<Map<String, Foo>>() {});
 ----
 
+
 [[polled-errors]]
 ===== Handling Errors
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -458,8 +458,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				throw (ProvisioningException) e;
 			}
 			else {
-				throw new BinderException("Exception thrown while starting consumer: ",
-						e);
+				throw new BinderException("Exception thrown while starting consumer: ", e);
 			}
 		}
 	}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java

@@ -17,8 +17,11 @@
 package org.springframework.cloud.stream.function;
 
 import org.springframework.beans.factory.FactoryBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.binder.PollableMessageSource;
 import org.springframework.cloud.stream.binding.BindableProxyFactory;
 import org.springframework.cloud.stream.binding.BoundTargetHolder;
+import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.util.Assert;
@@ -49,16 +52,26 @@ class BindableFunctionProxyFactory extends BindableProxyFactory {
 
 	private final StreamFunctionProperties functionProperties;
 
+	private final boolean pollable;
+
+	@Autowired
+	private GenericApplicationContext context;
+
 
 	BindableFunctionProxyFactory(String functionDefinition, int inputCount, int outputCount, StreamFunctionProperties functionProperties) {
+		this(functionDefinition, inputCount, outputCount, functionProperties, false);
+	}
+
+	BindableFunctionProxyFactory(String functionDefinition, int inputCount, int outputCount, StreamFunctionProperties functionProperties,
+			boolean pollable) {
 		super(null);
 		this.inputCount = inputCount;
 		this.outputCount = outputCount;
 		this.functionDefinition = functionDefinition;
 		this.functionProperties = functionProperties;
+		this.pollable = pollable;
 	}
 
-
 	@Override
 	public void afterPropertiesSet() {
 		Assert.notEmpty(BindableFunctionProxyFactory.this.bindingTargetFactories,
@@ -133,9 +146,18 @@ class BindableFunctionProxyFactory extends BindableProxyFactory {
 		if (this.functionProperties.getBindings().containsKey(name)) {
 			name = this.functionProperties.getBindings().get(name);
 		}
-		this.inputHolders.put(name,
-				new BoundTargetHolder(getBindingTargetFactory(SubscribableChannel.class)
-						.createInput(name), true));
+		if (this.pollable) {
+			PollableMessageSource pollableSource = (PollableMessageSource) getBindingTargetFactory(PollableMessageSource.class).createInput(name);
+			if (context != null && !context.containsBean(name)) {
+				context.registerBean(name, PollableMessageSource.class, () -> pollableSource);
+			}
+			this.inputHolders.put(name, new BoundTargetHolder(pollableSource, true));
+		}
+		else {
+			this.inputHolders.put(name,
+					new BoundTargetHolder(getBindingTargetFactory(SubscribableChannel.class)
+							.createInput(name), true));
+		}
 	}
 
 	private void createOutput(String name) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java

@@ -43,6 +43,8 @@ import reactor.util.function.Tuples;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.beans.factory.support.RootBeanDefinition;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
@@ -132,6 +134,11 @@ public class FunctionConfiguration {
 		return new FunctionBindingRegistrar(functionCatalog, streamFunctionProperties);
 	}
 
+	@Bean
+	public BeanFactoryPostProcessor po(Environment environment) {
+		return new PollableSourceRegistrar(environment);
+	}
+
 	@Bean
 	public InitializingBean functionInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
 			StreamFunctionProperties functionProperties, @Nullable BindableProxyFactory[] bindableProxyFactories,
@@ -617,6 +624,31 @@ public class FunctionConfiguration {
 		}
 	}
 
+	private static class PollableSourceRegistrar implements BeanFactoryPostProcessor {
+		private final Environment environment;
+
+		PollableSourceRegistrar(Environment environment) {
+			this.environment = environment;
+		}
+
+		@Override
+		public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
+			if (StringUtils.hasText(this.environment.getProperty("spring.cloud.stream.pollable-source"))) {
+				String[] sourceNames = this.environment.getProperty("spring.cloud.stream.pollable-source").split(";");
+
+				for (String sourceName : sourceNames) {
+					RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class);
+					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(sourceName);
+					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(1);
+					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(0);
+					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(new StreamFunctionProperties());
+					functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(true);
+					((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(sourceName + "_binding", functionBindableProxyDefinition);
+				}
+			}
+		}
+	}
+
 	/**
 	 * Creates and registers instances of BindableFunctionProxyFactory for each user defined function
 	 * thus triggering destination bindings between function arguments and destinations.
@@ -679,6 +711,7 @@ public class FunctionConfiguration {
 						}
 					}
 				}
+
 				if (StringUtils.hasText(this.environment.getProperty(SOURCE_PROPERY))) {
 					String[] sourceNames = this.environment.getProperty(SOURCE_PROPERY).split(";");
 
@@ -693,7 +726,6 @@ public class FunctionConfiguration {
 						}
 					}
 				}
-
 			}
 			else {
 				logger.info("Functional binding is disabled due to the presense of @EnableBinding annotation in your configuration");

spring-cloud-stream/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -17,6 +17,12 @@
 			"name": "spring.cloud.stream.sendto.destination",
 			"description": "The name of the header used to determine the name of the output destination",
 			"type": "java.lang.String"
+		},
+		{
+			"defaultValue": "none",
+			"name": "spring.cloud.stream.pollable-source",
+			"description": "A semi-colon delimited list of binding names of pollable sources. Binding names follow the same naming convention as functions. For example, name '...pollable-source=foobar' will be accessible as 'foobar-iin-0'' binding",
+			"type": "java.lang.String"
 		}
 	]
 }

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/PollableSourceTests.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2020-2020 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.function;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.WebApplicationType;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
+import org.springframework.cloud.stream.binder.DefaultPollableMessageSource;
+import org.springframework.cloud.stream.binder.PollableMessageSource;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Configuration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ * @author Oleg Zhurakousky
+ *
+ */
+public class PollableSourceTests {
+
+	@Test
+	public void test() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(PollableAppSampleConfiguration.class))
+						.web(WebApplicationType.NONE).run(
+								"--spring.jmx.enabled=false",
+								"--spring.cloud.stream.pollable-source=blah")) {
+
+
+			DefaultPollableMessageSource pollableSource = (DefaultPollableMessageSource) context.getBean(PollableMessageSource.class);
+			pollableSource.poll(message -> {
+				assertThat(message.getPayload()).isNotNull();
+			});
+		}
+	}
+
+	@EnableAutoConfiguration
+	@Configuration
+	public static class PollableAppSampleConfiguration {
+	}
+}