From e552cc75d66537e93be62232849c0f9d3d3d24f9 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 7 Mar 2019 10:08:49 +0100 Subject: [PATCH] GH-1590 Added support for implicit function bindings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Added support for implicit function binding where you no longer required to use EnableBinding making your application simply a boot app with stream/function bits coming being auto configured. - Added support for binding function without it being specified via ‘spring.cloud.strea.function.definition’ property if such function is unique (single per catalog) Resolves #1590 --- .../config/BinderFactoryConfiguration.java | 4 +- .../stream/config/BindingBeansRegistrar.java | 2 +- .../ImplicitFunctionBindingConfiguration.java | 101 +++++++++++ .../function/FunctionConfiguration.java | 55 +++--- .../GreenfieldFunctionEnableBindingTests.java | 8 +- .../ImplicitFunctionBindingTests.java | 169 ++++++++++++++++++ .../ProcessorToFunctionsSupportTests.java | 7 +- 7 files changed, 301 insertions(+), 45 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ImplicitFunctionBindingConfiguration.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java index 88be6dfa5..f6b2d1259 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2019 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. @@ -79,7 +79,7 @@ import org.springframework.validation.Validator; @Configuration @Role(BeanDefinition.ROLE_INFRASTRUCTURE) @EnableConfigurationProperties({ BindingServiceProperties.class }) -@Import({ ContentTypeConfiguration.class }) +@Import({ ContentTypeConfiguration.class, ImplicitFunctionBindingConfiguration.class }) @Deprecated public class BinderFactoryConfiguration { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java index 43fba6b8f..c195bc371 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java @@ -30,6 +30,7 @@ import org.springframework.util.ClassUtils; * @author Marius Bogoevici * @author Dave Syer * @author Artem Bilan + * @author Oleg Zhurakousky */ public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar { @@ -56,5 +57,4 @@ public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar { EnableBinding.class, ClassUtils.resolveClassName(className, null)); return enableBinding.value(); } - } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ImplicitFunctionBindingConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ImplicitFunctionBindingConfiguration.java new file mode 100644 index 000000000..2f01a8c5c --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ImplicitFunctionBindingConfiguration.java @@ -0,0 +1,101 @@ +/* + * Copyright 2019-2019 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 + * + * http://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.config; + +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.config.BeanFactoryPostProcessor; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.FunctionRegistry; +import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.context.annotation.Bean; +import org.springframework.core.env.Environment; +import org.springframework.core.env.StandardEnvironment; +import org.springframework.lang.Nullable; +import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; + +/** + * + * @author Oleg Zhurakousky + * @since 2.2 + * + */ +class ImplicitFunctionBindingConfiguration { + + @Bean + public BeanFactoryPostProcessor implicitFunctionBinder(Environment environment, + @Nullable FunctionRegistry functionCatalog, @Nullable FunctionInspector inspector) { + return new BeanFactoryPostProcessor() { + @Override + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { + if (functionCatalog != null && ObjectUtils.isEmpty(beanFactory.getBeanNamesForAnnotation(EnableBinding.class))) { + BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; + String name = determineFunctionName(functionCatalog, environment); + if (StringUtils.hasText(name)) { + Object definedFunction = functionCatalog.lookup(name); + Class inputType = inspector.getInputType(definedFunction); + Class outputType = inspector.getOutputType(definedFunction); + if (Void.class.isAssignableFrom(outputType)) { + bind(Sink.class, registry); + } + else if (Void.class.isAssignableFrom(inputType)) { + bind(Source.class, registry); + } + else { + bind(Processor.class, registry); + } + } + } + } + }; + } + + private void bind(Class type, BeanDefinitionRegistry registry) { + if (!registry.containsBeanDefinition(type.getName())) { + BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(type, type.getName(), registry); + BindingBeanDefinitionRegistryUtils.registerBindingTargetsQualifiedBeanDefinitions(type, type, registry); + } + } + + private String determineFunctionName(FunctionCatalog catalog, Environment environment) { + String name = environment.getProperty("spring.cloud.stream.function.definition"); + if (!StringUtils.hasText(name)) { + name = catalog.size() == 1 + ? catalog.getNames(Function.class).stream().findFirst() + .orElseGet(() -> catalog.getNames(Consumer.class).stream().findFirst() + .orElseGet(() -> catalog.getNames(Supplier.class).stream().findFirst().orElse(null))) + : null; + if (StringUtils.hasText(name)) { + ((StandardEnvironment) environment).getSystemProperties() + .putIfAbsent("spring.cloud.stream.function.definition", name); + } + } + + return 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 1407d9a37..3125d5371 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 @@ -20,12 +20,13 @@ import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; -import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.stream.config.BinderFactoryConfiguration; +import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.cloud.stream.messaging.Processor; @@ -33,8 +34,10 @@ import org.springframework.cloud.stream.messaging.Sink; import org.springframework.cloud.stream.messaging.Source; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.lang.Nullable; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.SubscribableChannel; @@ -45,20 +48,11 @@ import org.springframework.messaging.SubscribableChannel; * @since 2.1 */ @Configuration -@ConditionalOnProperty("spring.cloud.stream.function.definition") @EnableConfigurationProperties(StreamFunctionProperties.class) +@Import(BinderFactoryConfiguration.class) +@AutoConfigureBefore(BindingServiceConfiguration.class) public class FunctionConfiguration { - - @Autowired(required = false) - private Source source; - - @Autowired(required = false) - private Processor processor; - - @Autowired(required = false) - private Sink sink; - @Bean public IntegrationFlowFunctionSupport functionSupport( FunctionCatalog functionCatalog, FunctionInspector functionInspector, @@ -69,11 +63,6 @@ public class FunctionConfiguration { messageConverterFactory, functionProperties, bindingServiceProperties); } -// @Bean -// public FunctionCatalogWrapper functionCatalogWrapper(FunctionCatalog catalog) { -// return new FunctionCatalogWrapper(catalog); -// } - /** * This configuration creates an instance of the {@link IntegrationFlow} from standard * Spring Cloud Stream bindings such as {@link Source}, {@link Processor} and @@ -88,42 +77,46 @@ public class FunctionConfiguration { * the IntegrationFlow that may have been already defined by the existing (extended) * app. * @param functionSupport support for registering beans + * @param source source binding + * @param processor processor binding + * @param sink sink binding * @return integration flow for Stream */ @ConditionalOnMissingBean @Bean public IntegrationFlow integrationFlowCreator( - IntegrationFlowFunctionSupport functionSupport) { + IntegrationFlowFunctionSupport functionSupport, + @Nullable Source source, @Nullable Processor processor, @Nullable Sink sink) { if (functionSupport.containsFunction(Consumer.class) - && consumerBindingPresent()) { + && consumerBindingPresent(processor, sink)) { return functionSupport - .integrationFlowForFunction(getInputChannel(), getOutputChannel()) + .integrationFlowForFunction(getInputChannel(processor, sink), getOutputChannel(processor, source)) .get(); } else if (functionSupport.containsFunction(Function.class) - && consumerBindingPresent()) { + && consumerBindingPresent(processor, sink)) { return functionSupport - .integrationFlowForFunction(getInputChannel(), getOutputChannel()) + .integrationFlowForFunction(getInputChannel(processor, sink), getOutputChannel(processor, source)) .get(); } else if (functionSupport.containsFunction(Supplier.class)) { return functionSupport.integrationFlowFromNamedSupplier() - .channel(getOutputChannel()).get(); + .channel(getOutputChannel(processor, source)).get(); } return null; } - private boolean consumerBindingPresent() { - return this.processor != null || this.sink != null; + private boolean consumerBindingPresent(Processor processor, Sink sink) { + return processor != null || sink != null; } - private SubscribableChannel getInputChannel() { - return this.processor != null ? this.processor.input() : this.sink.input(); + private SubscribableChannel getInputChannel(Processor processor, Sink sink) { + return processor != null ? processor.input() : sink.input(); } - private MessageChannel getOutputChannel() { - return this.processor != null ? this.processor.output() - : (this.source != null ? this.source.output() : new NullChannel()); + private MessageChannel getOutputChannel(Processor processor, Source source) { + return processor != null ? processor.output() + : (source != null ? source.output() : new NullChannel()); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java index f04f1ba46..6d210c868 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java @@ -26,7 +26,6 @@ import java.util.function.Supplier; import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.Test; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; @@ -208,21 +207,18 @@ public class GreenfieldFunctionEnableBindingTests { @EnableBinding(Source.class) public static class HttpInboundEndpoint { - @Autowired - private Source source; - @Bean public Function upperCase() { return String::toUpperCase; } @Bean - public HttpRequestHandlingEndpointSupport doFoo() { + public HttpRequestHandlingEndpointSupport doFoo(Source source) { HttpRequestHandlerEndpointSpec httpRequestHandler = Http .inboundChannelAdapter("/*") .requestMapping(requestMapping -> requestMapping .methods(HttpMethod.POST).consumes("*/*")) - .requestChannel(this.source.output()); + .requestChannel(source.output()); return httpRequestHandler.get(); } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java new file mode 100644 index 000000000..c58d8fc49 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java @@ -0,0 +1,169 @@ +/* + * Copyright 2019-2019 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 + * + * http://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 java.util.function.Consumer; +import java.util.function.Function; + +import org.junit.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * + * @author Oleg Zhurakousky + * + */ +public class ImplicitFunctionBindingTests { + + @Test + public void testBindingWithNoEnableBindingConfiguration() { + + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + NoEnableBindingConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder + .withPayload("Hello".getBytes()).build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage.getPayload()).isEqualTo("Hello".getBytes()); + + } + } + + @Test + public void testBindingWithEnableBindingConfiguration() { + + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + EnableBindingConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder + .withPayload("Hello".getBytes()).build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage.getPayload()).isEqualTo("Hello".getBytes()); + + } + } + + @Test + public void testBindingWithNoEnableBindingAndNoDefinitionPropertyConfiguration() { + + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SingleFunctionConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder + .withPayload("Hello".getBytes()).build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage.getPayload()).isEqualTo("Hello".getBytes()); + + } + } + + @EnableAutoConfiguration + public static class NoEnableBindingConfiguration { + + @Bean + public Function func() { + return x -> { + System.out.println("Function"); + return x; + }; + } + + @Bean + public Consumer cons() { + return x -> { + System.out.println("Consumer"); + }; + } + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class EnableBindingConfiguration { + + @Bean + public Function func() { + return x -> { + System.out.println("Function"); + return x; + }; + } + + @Bean + public Consumer cons() { + return x -> { + System.out.println("Consumer"); + }; + } + } + + @EnableAutoConfiguration + public static class SingleFunctionConfiguration { + + @Bean + public Function func() { + return x -> { + System.out.println("Function"); + return x; + }; + } + + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java index 3ee708a19..d69637be3 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java @@ -144,13 +144,10 @@ public class ProcessorToFunctionsSupportTests { @Import(BaseProcessorConfiguration.class) public static class ConsumerConfiguration { - @Autowired - OutputDestination out; - @Bean - public Consumer log() { + public Consumer log(OutputDestination out) { return x -> { - DirectFieldAccessor dfa = new DirectFieldAccessor(this.out); + DirectFieldAccessor dfa = new DirectFieldAccessor(out); MessageChannel channel = (MessageChannel) dfa.getPropertyValue("channel"); channel.send(new GenericMessage(x.getBytes())); };