diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java index 250ef4be1..46c20c8bd 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 the original author or authors. + * Copyright 2019-2021 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. @@ -26,7 +26,8 @@ import java.util.Set; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.internal.InternalPropertyNames; @@ -48,7 +49,6 @@ public class AbstractBindableProxyFactory implements Bindable { @Value("${" + InternalPropertyNames.NAMESPACE_PROPERTY_NAME + ":}") private String namespace; - @Autowired protected Map bindingTargetFactories; protected Map inputHolders = new LinkedHashMap<>(); @@ -57,10 +57,16 @@ public class AbstractBindableProxyFactory implements Bindable { protected Class type; + private BeanFactory beanFactory; + public AbstractBindableProxyFactory(Class type) { this.type = type; } + protected void populateBindingTargetFactories(BeanFactory beanFactory) { + this.bindingTargetFactories = ((ListableBeanFactory) beanFactory).getBeansOfType(BindingTargetFactory.class); + } + protected BindingTargetFactory getBindingTargetFactory(Class bindingTargetType) { List candidateBindingTargetFactories = new ArrayList<>(); for (Map.Entry bindingTargetFactoryEntry : this.bindingTargetFactories diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java index 08aec34f3..66d255f2d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2021 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. @@ -26,6 +26,9 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.aop.framework.ProxyFactory; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.cloud.stream.annotation.EnableBinding; @@ -45,10 +48,11 @@ import org.springframework.util.ReflectionUtils; * @author David Syer * @author Ilayaperumal Gopinathan * @author Oleg Zhurakousky + * @author Soby Chacko * @see EnableBinding */ public class BindableProxyFactory extends AbstractBindableProxyFactory - implements MethodInterceptor, FactoryBean, InitializingBean { + implements MethodInterceptor, FactoryBean, InitializingBean, BeanFactoryAware { private static Log log = LogFactory.getLog(BindableProxyFactory.class); @@ -56,6 +60,8 @@ public class BindableProxyFactory extends AbstractBindableProxyFactory private Object proxy; + private BeanFactory beanFactory; + public BindableProxyFactory(Class type) { super(type); this.type = type; @@ -112,6 +118,7 @@ public class BindableProxyFactory extends AbstractBindableProxyFactory @Override public void afterPropertiesSet() { + populateBindingTargetFactories(this.beanFactory); Assert.notEmpty(BindableProxyFactory.this.bindingTargetFactories, "'bindingTargetFactories' cannot be empty"); ReflectionUtils.doWithMethods(this.type, method -> { @@ -159,4 +166,8 @@ public class BindableProxyFactory extends AbstractBindableProxyFactory return true; } + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java index dd68be71a..642333bdf 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2019 the original author or authors. + * Copyright 2015-2021 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. @@ -170,6 +170,7 @@ public class BinderFactoryAutoConfiguration { ClassLoader classLoader = configurableApplicationContext.getClassLoader(); try { Enumeration resources = classLoader.getResources("META-INF/spring.binders"); + // see if test binder is available on the classpath and if so add it to the binderTypes try { BinderType bt = new BinderType("integration", new Class[] { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 00f499b97..c7153cdae 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2021 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. @@ -70,6 +70,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.DestinationResolver; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; /** @@ -198,7 +199,6 @@ public class BindingServiceConfiguration { public BinderFactory binderFactory(BinderTypeRegistry binderTypeRegistry, BindingServiceProperties bindingServiceProperties, ObjectProvider binderCustomizerProvider) { - DefaultBinderFactory binderFactory = new DefaultBinderFactory( getBinderConfigurations(binderTypeRegistry, bindingServiceProperties), binderTypeRegistry, binderCustomizerProvider.getIfUnique()); @@ -271,11 +271,12 @@ public class BindingServiceConfiguration { @Bean @ConditionalOnMissingBean public BinderAwareRouter binderAwareRouterBeanPostProcessor( - @Autowired(required = false) AbstractMappingMessageRouter[] routers, + @Autowired(required = false) List routers, @Autowired(required = false) @Qualifier("binderAwareChannelResolver") DestinationResolver channelResolver) { - - return new BinderAwareRouter(routers, channelResolver); + final AbstractMappingMessageRouter[] routersArray = CollectionUtils.isEmpty(routers) ? + new AbstractMappingMessageRouter[]{} : routers.toArray(new AbstractMappingMessageRouter[]{}); + return new BinderAwareRouter(routersArray, channelResolver); } @Bean diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java index 12161a19b..3049920ec 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 the original author or authors. + * Copyright 2019-2021 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. @@ -39,6 +39,7 @@ import org.springframework.util.CollectionUtils; * function with the name `myFunction`. * * @author Oleg Zhurakousky + * @author Soby Chacko * * @since 3.0 */ @@ -74,6 +75,7 @@ class BindableFunctionProxyFactory extends BindableProxyFactory { @Override public void afterPropertiesSet() { + populateBindingTargetFactories(context.getBeanFactory()); Assert.notEmpty(BindableFunctionProxyFactory.this.bindingTargetFactories, "'bindingTargetFactories' cannot be empty"); 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 3f97512a1..ea50c7092 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 @@ -1,5 +1,5 @@ /* - * Copyright 2018-2020 the original author or authors. + * Copyright 2018-2021 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. @@ -116,6 +116,7 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky * @author David Turanski * @author Ilayaperumal Gopinathan + * @author Soby Chacko * @since 2.1 */ @Configuration @@ -147,10 +148,10 @@ public class FunctionConfiguration { } @Bean - public InitializingBean functionInitializer(FunctionCatalog functionCatalog, FunctionInspector functionInspector, - StreamFunctionProperties functionProperties, @Nullable BindableProxyFactory[] bindableProxyFactories, - BindingServiceProperties serviceProperties, ConfigurableApplicationContext applicationContext, - FunctionBindingRegistrar bindingHolder, StreamBridge streamBridge) { + public InitializingBean functionInitializer(FunctionCatalog functionCatalog, + StreamFunctionProperties functionProperties, + BindingServiceProperties serviceProperties, ConfigurableApplicationContext applicationContext, + StreamBridge streamBridge) { boolean shouldCreateInitializer = applicationContext.containsBean("output") || ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class)); @@ -167,10 +168,10 @@ public class FunctionConfiguration { @Bean InitializingBean supplierInitializer(FunctionCatalog functionCatalog, StreamFunctionProperties functionProperties, GenericApplicationContext context, BindingServiceProperties serviceProperties, - @Nullable BindableFunctionProxyFactory[] proxyFactories, StreamBridge streamBridge, + @Nullable List proxyFactories, StreamBridge streamBridge, TaskScheduler taskScheduler) { - if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class)) || proxyFactories == null) { + if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class)) || CollectionUtils.isEmpty(proxyFactories)) { return null; }