From 9a906e7369768bd47c28325b22752818adb5680b Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 22 Sep 2020 20:32:06 +0200 Subject: [PATCH] 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 --- .../main/asciidoc/spring-cloud-stream.adoc | 20 +++---- .../binder/AbstractMessageChannelBinder.java | 3 +- .../BindableFunctionProxyFactory.java | 30 ++++++++-- .../function/FunctionConfiguration.java | 34 ++++++++++- ...itional-spring-configuration-metadata.json | 6 ++ .../stream/function/PollableSourceTests.java | 59 +++++++++++++++++++ 6 files changed, 134 insertions(+), 18 deletions(-) create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/PollableSourceTests.java diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index b00cd45bf..750381639 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/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>() {}); ---- + [[polled-errors]] ===== Handling Errors diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index ab766f7be..5dcd9b518 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -458,8 +458,7 @@ public abstract class AbstractMessageChannelBinder 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) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index fe70fe6c9..09e9352f9 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/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"); diff --git a/spring-cloud-stream/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/spring-cloud-stream/src/main/resources/META-INF/additional-spring-configuration-metadata.json index a398e4483..c8cc7ac9c 100644 --- a/spring-cloud-stream/src/main/resources/META-INF/additional-spring-configuration-metadata.json +++ b/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" } ] } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/PollableSourceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/PollableSourceTests.java new file mode 100644 index 000000000..679ba886c --- /dev/null +++ b/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 { + } +}