From 80e1e37a39f1de7f2f1742ffa74b48226c2741a8 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 5 Oct 2022 15:12:16 +0200 Subject: [PATCH] GH-2522 Add support for propagating select beans across child AC Resolves #2522 --- .../stream/binder/DefaultBinderFactory.java | 57 +++++++++++++++++-- .../src/main/resources/META-INF/shared.beans | 12 ++++ 2 files changed, 63 insertions(+), 6 deletions(-) create mode 100644 core/spring-cloud-stream/src/main/resources/META-INF/shared.beans diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java index ab908b97a..8786a0331 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -16,16 +16,19 @@ package org.springframework.cloud.stream.binder; +import java.net.URL; import java.util.AbstractMap.SimpleImmutableEntry; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.Enumeration; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Properties; import java.util.Set; import java.util.stream.Stream; @@ -36,8 +39,6 @@ import org.springframework.beans.factory.DisposableBean; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry; import org.springframework.cloud.stream.config.ListenerContainerCustomizer; -import org.springframework.cloud.stream.config.SpelExpressionConverterConfiguration; -import org.springframework.cloud.stream.config.SpelExpressionConverterConfiguration.SpelConverter; import org.springframework.cloud.stream.reflection.GenericsUtils; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; @@ -47,13 +48,17 @@ import org.springframework.context.ApplicationListener; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.support.GenericApplicationContext; +import org.springframework.core.convert.converter.Converter; import org.springframework.core.convert.support.GenericConversionService; import org.springframework.core.env.ConfigurableEnvironment; import org.springframework.core.env.MapPropertySource; +import org.springframework.core.io.UrlResource; +import org.springframework.core.io.support.PropertiesLoaderUtils; import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.converter.MessageConverter; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -377,13 +382,10 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl AnnotationConfigApplicationContext binderProducingContext = new AnnotationConfigApplicationContext(); if (this.context != null) { binderProducingContext.getBeanFactory().setConversionService(this.context.getBeanFactory().getConversionService()); - GenericConversionService cs = (GenericConversionService) ((GenericApplicationContext) binderProducingContext).getBeanFactory().getConversionService(); - SpelConverter spelConverter = new SpelConverter(); - cs.addConverter(spelConverter); } + List sourceClasses = new ArrayList<>(); sourceClasses.addAll(Arrays.asList(binderType.getConfigurationClasses())); - sourceClasses.addAll(Collections.singletonList(SpelExpressionConverterConfiguration.class)); if (binderProperties.containsKey("spring.main.sources")) { String sources = (String) binderProperties.get("spring.main.sources"); if (StringUtils.hasText(sources)) { @@ -410,6 +412,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl binderProducingContext.setParent(this.context); } else if (this.context != null) { + this.propagateSharedBeans(binderProducingContext); Map customizers = this.context.getBeansOfType(ListenerContainerCustomizer.class); if (!CollectionUtils.isEmpty(customizers)) { for (Entry customizerEntry : customizers.entrySet()) { @@ -467,6 +470,48 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl return binderProducingContext; } + private void propagateSharedBeans(GenericApplicationContext binderProducingContext) { + GenericConversionService binderProducingConversionService = (GenericConversionService) binderProducingContext.getBeanFactory().getConversionService(); + try { + Enumeration resources = ClassUtils.getDefaultClassLoader().getResources("META-INF/shared.beans"); + while (resources.hasMoreElements()) { + URL url = (URL) resources.nextElement(); + UrlResource resource = new UrlResource(url); + Properties properties = PropertiesLoaderUtils.loadProperties(resource); + Set classNames = properties.keySet(); + for (Object className : classNames) { + Class beanType = this.loadClass(((String) className).trim()); + if (beanType != null) { + Map beansOfType = this.context.getBeansOfType(beanType); + beansOfType.entrySet().stream().forEach(entry -> { + Object bean = entry.getValue(); + if (bean instanceof Converter) { + binderProducingConversionService.addConverter((Converter) bean); + } + else { + binderProducingContext.registerBean(entry.getKey() + "_child", beanType, () -> entry.getValue()); + } + }); + } + } + } + } + catch (Exception e) { + logger.warn("Failed to propagate child beans. This may cause issues in your application", e); + } + } + + @SuppressWarnings("unchecked") + private Class loadClass(String className) { + try { + return (Class) ClassUtils.getDefaultClassLoader().loadClass(((String) className).trim()); + } + catch (Exception e) { + logger.debug("Attempt to load " + className + " failed.", e); + return null; + } + } + /** * Creates a bare minimum application context that can be initialized by AOT. * diff --git a/core/spring-cloud-stream/src/main/resources/META-INF/shared.beans b/core/spring-cloud-stream/src/main/resources/META-INF/shared.beans new file mode 100644 index 000000000..93310e0ad --- /dev/null +++ b/core/spring-cloud-stream/src/main/resources/META-INF/shared.beans @@ -0,0 +1,12 @@ +org.springframework.boot.autoconfigure.amqp.ConnectionFactoryCustomizer +org.springframework.cloud.stream.config.SpelExpressionConverterConfiguration$SpelConverter +org.springframework.cloud.stream.config.ListenerContainerCustomizer +org.springframework.cloud.stream.binder.kafka.ListenerContainerWithDlqAndRetryCustomizer +org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer +org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer +org.springframework.cloud.stream.binder.kafka.provisioning.AdminClientConfigCustomizer +org.springframework.boot.autoconfigure.kafka.StreamsBuilderFactoryBeanCustomizer +org.springframework.kafka.config.KafkaStreamsCustomizer +org.springframework.rabbit.stream.listener.ConsumerCustomizer +org.springframework.amqp.core.DeclarableCustomizer +org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer