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 f6b2d1259..11076e399 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 @@ -31,26 +31,39 @@ import java.util.Properties; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Value; 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.boot.context.properties.EnableConfigurationProperties; +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.binder.BinderType; import org.springframework.cloud.stream.binder.BinderTypeRegistry; import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry; +import org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils; import org.springframework.cloud.stream.binding.CompositeMessageChannelConfigurer; import org.springframework.cloud.stream.binding.MessageChannelConfigurer; import org.springframework.cloud.stream.binding.MessageConverterConfigurer; import org.springframework.cloud.stream.binding.MessageSourceBindingTargetFactory; import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +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.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Role; +import org.springframework.core.env.Environment; +import org.springframework.core.env.StandardEnvironment; import org.springframework.core.io.Resource; import org.springframework.core.io.UrlResource; import org.springframework.core.io.support.PropertiesLoaderUtils; @@ -63,6 +76,7 @@ import org.springframework.messaging.handler.annotation.support.HeadersMethodArg import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver; import org.springframework.util.ClassUtils; +import org.springframework.util.ObjectUtils; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; import org.springframework.validation.Validator; @@ -79,7 +93,7 @@ import org.springframework.validation.Validator; @Configuration @Role(BeanDefinition.ROLE_INFRASTRUCTURE) @EnableConfigurationProperties({ BindingServiceProperties.class }) -@Import({ ContentTypeConfiguration.class, ImplicitFunctionBindingConfiguration.class }) +@Import(ContentTypeConfiguration.class) @Deprecated public class BinderFactoryConfiguration { @@ -113,6 +127,8 @@ public class BinderFactoryConfiguration { return parsedBinderConfigurations; } + + @Bean(IntegrationContextUtils.MESSAGE_HANDLER_FACTORY_BEAN_NAME) public static MessageHandlerMethodFactory messageHandlerMethodFactory( CompositeMessageConverterFactory compositeMessageConverterFactory, @@ -225,4 +241,52 @@ public class BinderFactoryConfiguration { return new CompositeMessageChannelConfigurer(configurerList); } + @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 String determineFunctionName(FunctionCatalog catalog, Environment environment) { + String name = environment.getProperty("spring.cloud.stream.function.definition"); + if (!StringUtils.hasText(name) && catalog.size() == 1) { + name = ((FunctionInspector) catalog).getName(catalog.lookup("")); + if (StringUtils.hasText(name)) { + ((StandardEnvironment) environment).getSystemProperties() + .putIfAbsent("spring.cloud.stream.function.definition", name); + } + } + + return name; + } + + private void bind(Class type, BeanDefinitionRegistry registry) { + if (!registry.containsBeanDefinition(type.getName())) { + BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(type, type.getName(), registry); + BindingBeanDefinitionRegistryUtils.registerBindingTargetsQualifiedBeanDefinitions(type, type, registry); + } + } + } 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 deleted file mode 100644 index 2f01a8c5c..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ImplicitFunctionBindingConfiguration.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * 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; - } -}