Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-8611: Extract MessagingAnnotationBeanPostProcessor (#8661)

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/8611

According to the latest Spring Framework requirements related to AOT the
`BeanDefinitionRegistryPostProcessor` must not do anything but only bean registrations

* Extract a `MessagingAnnotationBeanPostProcessor` component with a `BeanPostProcessor` logic
from the `MessagingAnnotationPostProcessor` and leave the last one as just
a `BeanDefinitionRegistryPostProcessor` impl
* Fix tests according to a new logic
This commit is contained in:
Artem Bilan
2023-06-28 08:59:37 -04:00
committed by GitHub
parent 13980af1e1
commit 4583afe0ad
7 changed files with 321 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files

248
spring-integration-core/src/main/java/org/springframework/integration/config/MessagingAnnotationBeanPostProcessor.java Normal file
View File

@@ -0,0 +1,248 @@
/*
* Copyright 2023 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.integration.config;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.annotation.Bean;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.MergedAnnotation;
import org.springframework.core.annotation.MergedAnnotations;
import org.springframework.integration.annotation.EndpointId;
import org.springframework.integration.annotation.Role;
import org.springframework.integration.config.annotation.MethodAnnotationPostProcessor;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.util.MessagingAnnotationUtils;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* An infrastructure {@link BeanPostProcessor} implementation that processes method-level
* messaging annotations such as @Transformer, @Splitter, @Router, and @Filter.
*
* @author Artem Bilan
*
* @since 6.2
*/
public class MessagingAnnotationBeanPostProcessor
implements BeanPostProcessor, BeanFactoryAware, SmartInitializingSingleton {
private final Set<Class<?>> noAnnotationsCache = Collections.newSetFromMap(new ConcurrentHashMap<>(256));
private final Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> postProcessors;
private final List<Runnable> methodsToPostProcessAfterContextInitialization = new ArrayList<>();
private final BeanDefinitionRegistry registry;
private ConfigurableListableBeanFactory beanFactory;
private volatile boolean initialized;
public MessagingAnnotationBeanPostProcessor(BeanDefinitionRegistry registry,
Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> postProcessors) {
this.registry = registry;
this.postProcessors = postProcessors;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = (ConfigurableListableBeanFactory) beanFactory;
}
@Override
public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
Assert.notNull(this.beanFactory, "BeanFactory must not be null");
Class<?> beanClass = AopUtils.getTargetClass(bean);
// the set will hold records of prior class scans and indicate if no messaging annotations were found
if (this.noAnnotationsCache.contains(beanClass)) {
return bean;
}
ReflectionUtils.doWithMethods(beanClass,
method -> doWithMethod(method, bean, beanName, beanClass),
ReflectionUtils.USER_DECLARED_METHODS);
return bean;
}
private void doWithMethod(Method method, Object bean, String beanName, Class<?> beanClass) {
MergedAnnotations mergedAnnotations = MergedAnnotations.from(method);
boolean noMessagingAnnotations = true;
// See MessagingAnnotationPostProcessor.postProcessBeanDefinitionRegistry(BeanDefinitionRegistry)
if (!mergedAnnotations.isPresent(Bean.class)) {
List<MessagingMetaAnnotation> messagingAnnotations =
obtainMessagingAnnotations(this.postProcessors.keySet(), mergedAnnotations,
method.toGenericString());
for (MessagingMetaAnnotation messagingAnnotation : messagingAnnotations) {
noMessagingAnnotations = false;
Class<? extends Annotation> annotationType = messagingAnnotation.annotationType;
List<Annotation> annotationChain =
MessagingAnnotationUtils.getAnnotationChain(messagingAnnotation.annotation, annotationType);
processAnnotationTypeOnMethod(bean, beanName, method, annotationType, annotationChain);
}
}
if (noMessagingAnnotations) {
this.noAnnotationsCache.add(beanClass);
}
}
private void processAnnotationTypeOnMethod(Object bean, String beanName, Method method,
Class<? extends Annotation> annotationType, List<Annotation> annotations) {
MethodAnnotationPostProcessor<?> postProcessor = this.postProcessors.get(annotationType);
if (postProcessor != null && postProcessor.supportsPojoMethod()
&& postProcessor.shouldCreateEndpoint(method, annotations)) {
Method targetMethod = method;
if (AopUtils.isJdkDynamicProxy(bean)) {
try {
targetMethod = bean.getClass().getMethod(method.getName(), method.getParameterTypes());
}
catch (NoSuchMethodException e) {
throw new IllegalArgumentException("Service methods must be extracted to the service "
+ "interface for JdkDynamicProxy. The affected bean is: '" + beanName + "' "
+ "and its method: '" + method + "'", e);
}
}
if (this.initialized) {
postProcessMethodAndRegisterEndpointIfAny(bean, beanName, method, annotationType, annotations,
postProcessor, targetMethod);
}
else {
Method methodToPostProcess = targetMethod;
this.methodsToPostProcessAfterContextInitialization.add(() ->
postProcessMethodAndRegisterEndpointIfAny(bean, beanName, method, annotationType, annotations,
postProcessor, methodToPostProcess));
}
}
}
@SuppressWarnings("unchecked")
private void postProcessMethodAndRegisterEndpointIfAny(Object bean, String beanName, Method method,
Class<? extends Annotation> annotationType, List<Annotation> annotations,
MethodAnnotationPostProcessor<?> postProcessor, Method targetMethod) {
Object result = postProcessor.postProcess(bean, beanName, targetMethod, annotations);
if (result instanceof AbstractEndpoint endpoint) {
String autoStartup = MessagingAnnotationUtils.resolveAttribute(annotations, "autoStartup", String.class);
if (StringUtils.hasText(autoStartup)) {
autoStartup = this.beanFactory.resolveEmbeddedValue(autoStartup);
if (StringUtils.hasText(autoStartup)) {
endpoint.setAutoStartup(Boolean.parseBoolean(autoStartup));
}
}
String phase = MessagingAnnotationUtils.resolveAttribute(annotations, "phase", String.class);
if (StringUtils.hasText(phase)) {
phase = this.beanFactory.resolveEmbeddedValue(phase);
if (StringUtils.hasText(phase)) {
endpoint.setPhase(Integer.parseInt(phase));
}
}
Role role = AnnotationUtils.findAnnotation(method, Role.class);
if (role != null) {
endpoint.setRole(role.value());
}
String endpointBeanName = generateBeanName(beanName, method, annotationType);
endpoint.setBeanName(endpointBeanName);
this.registry.registerBeanDefinition(endpointBeanName,
new RootBeanDefinition((Class<AbstractEndpoint>) endpoint.getClass(), () -> endpoint));
this.beanFactory.getBean(endpointBeanName);
}
}
protected String generateBeanName(String originalBeanName, Method method,
Class<? extends Annotation> annotationType) {
String name = MessagingAnnotationUtils.endpointIdValue(method);
if (!StringUtils.hasText(name)) {
String baseName = originalBeanName + "." + method.getName() + "."
+ ClassUtils.getShortNameAsProperty(annotationType);
name = baseName;
int count = 1;
while (this.beanFactory.containsBean(name)) {
name = baseName + "#" + (++count);
}
}
return name;
}
@Override
public void afterSingletonsInstantiated() {
this.initialized = true;
this.methodsToPostProcessAfterContextInitialization.forEach(Runnable::run);
this.methodsToPostProcessAfterContextInitialization.clear();
}
protected static List<MessagingMetaAnnotation> obtainMessagingAnnotations(
Set<Class<? extends Annotation>> postProcessors, MergedAnnotations annotations, String identified) {
List<MessagingMetaAnnotation> messagingAnnotations = new ArrayList<>();
for (Class<? extends Annotation> annotationType : postProcessors) {
annotations.stream()
.filter((ann) -> ann.getType().equals(annotationType))
.map(MergedAnnotation::getRoot)
.map(MergedAnnotation::synthesize)
.map((ann) -> new MessagingMetaAnnotation(ann, annotationType))
.forEach(messagingAnnotations::add);
}
if (annotations.get(EndpointId.class, (ann) -> ann.hasNonDefaultValue("value")).isPresent()
&& messagingAnnotations.size() > 1) {
throw new IllegalStateException("@EndpointId on " + identified
+ " can only have one EIP annotation, found: " + messagingAnnotations.size());
}
return messagingAnnotations;
}
public record MessagingMetaAnnotation(Annotation annotation, Class<? extends Annotation> annotationType) {
}
}

235
spring-integration-core/src/main/java/org/springframework/integration/config/MessagingAnnotationPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -16,58 +16,38 @@
package org.springframework.integration.config;
import java.beans.Introspector;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.aot.AotDetector;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.annotation.Bean;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.MergedAnnotation;
import org.springframework.core.annotation.MergedAnnotations;
import org.springframework.core.type.MethodMetadata;
import org.springframework.core.type.StandardMethodMetadata;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.BridgeFrom;
import org.springframework.integration.annotation.BridgeTo;
import org.springframework.integration.annotation.EndpointId;
import org.springframework.integration.annotation.Filter;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Role;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.config.annotation.MethodAnnotationPostProcessor;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.util.MessagingAnnotationUtils;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* A {@link BeanPostProcessor} implementation that processes method-level
@@ -79,26 +59,18 @@ import org.springframework.util.StringUtils;
* @author Gary Russell
* @author Rick Hogge
*/
public class MessagingAnnotationPostProcessor
implements BeanDefinitionRegistryPostProcessor, BeanPostProcessor, SmartInitializingSingleton {
protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR
public class MessagingAnnotationPostProcessor implements BeanDefinitionRegistryPostProcessor {
private final Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> postProcessors = new HashMap<>();
private final Set<Class<?>> noAnnotationsCache = Collections.newSetFromMap(new ConcurrentHashMap<>(256));
private final List<Runnable> methodsToPostProcessAfterContextInitialization = new ArrayList<>();
private BeanDefinitionRegistry registry;
private ConfigurableListableBeanFactory beanFactory;
private volatile boolean initialized;
@Override
public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
this.registry = registry;
this.postProcessors.put(Filter.class, new FilterAnnotationPostProcessor());
this.postProcessors.put(Router.class, new RouterAnnotationPostProcessor());
this.postProcessors.put(Transformer.class, new TransformerAnnotationPostProcessor());
@@ -108,6 +80,7 @@ public class MessagingAnnotationPostProcessor
this.postProcessors.put(InboundChannelAdapter.class, new InboundChannelAdapterAnnotationPostProcessor());
this.postProcessors.put(BridgeFrom.class, new BridgeFromAnnotationPostProcessor());
this.postProcessors.put(BridgeTo.class, new BridgeToAnnotationPostProcessor());
Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> customPostProcessors =
setupCustomPostProcessors();
if (!CollectionUtils.isEmpty(customPostProcessors)) {
@@ -118,16 +91,22 @@ public class MessagingAnnotationPostProcessor
.map(BeanFactoryAware.class::cast)
.forEach((processor) -> processor.setBeanFactory((BeanFactory) this.registry));
if (!AotDetector.useGeneratedArtifacts()) {
String[] beanNames = registry.getBeanDefinitionNames();
this.registry.registerBeanDefinition(
Introspector.decapitalize(MessagingAnnotationBeanPostProcessor.class.getName()),
BeanDefinitionBuilder.rootBeanDefinition(MessagingAnnotationBeanPostProcessor.class)
.setRole(BeanDefinition.ROLE_INFRASTRUCTURE)
.addConstructorArgValue(this.registry)
.addConstructorArgValue(this.postProcessors)
.getBeanDefinition());
for (String beanName : beanNames) {
BeanDefinition beanDef = registry.getBeanDefinition(beanName);
if (beanDef instanceof AnnotatedBeanDefinition annotatedBeanDefinition
&& annotatedBeanDefinition.getFactoryMethodMetadata() != null) {
String[] beanNames = registry.getBeanDefinitionNames();
processCandidate(beanName, annotatedBeanDefinition);
}
for (String beanName : beanNames) {
BeanDefinition beanDef = registry.getBeanDefinition(beanName);
if (beanDef instanceof AnnotatedBeanDefinition annotatedBeanDefinition
&& annotatedBeanDefinition.getFactoryMethodMetadata() != null) {
processCandidate(beanName, annotatedBeanDefinition);
}
}
}
@@ -139,39 +118,19 @@ public class MessagingAnnotationPostProcessor
annotations = MergedAnnotations.from(standardMethodMetadata.getIntrospectedMethod());
}
List<MessagingMetaAnnotation> messagingAnnotations = obtainMessagingAnnotations(annotations, beanName);
List<MessagingAnnotationBeanPostProcessor.MessagingMetaAnnotation> messagingAnnotations =
MessagingAnnotationBeanPostProcessor.obtainMessagingAnnotations(this.postProcessors.keySet(),
annotations, beanName);
for (MessagingMetaAnnotation messagingAnnotation : messagingAnnotations) {
Class<? extends Annotation> annotationType = messagingAnnotation.annotationType;
for (MessagingAnnotationBeanPostProcessor.MessagingMetaAnnotation messagingAnnotation : messagingAnnotations) {
Class<? extends Annotation> annotationType = messagingAnnotation.annotationType();
List<Annotation> annotationChain =
MessagingAnnotationUtils.getAnnotationChain(messagingAnnotation.annotation, annotationType);
MessagingAnnotationUtils.getAnnotationChain(messagingAnnotation.annotation(), annotationType);
processMessagingAnnotationOnBean(beanName, beanDefinition, annotationType, annotationChain);
}
}
private List<MessagingMetaAnnotation> obtainMessagingAnnotations(MergedAnnotations annotations, String identified) {
List<MessagingMetaAnnotation> messagingAnnotations = new ArrayList<>();
for (Class<? extends Annotation> annotationType : this.postProcessors.keySet()) {
annotations.stream()
.filter((ann) -> ann.getType().equals(annotationType))
.map(MergedAnnotation::getRoot)
.map(MergedAnnotation::synthesize)
.map((ann) -> new MessagingMetaAnnotation(ann, annotationType))
.forEach(messagingAnnotations::add);
}
if (annotations.get(EndpointId.class, (ann) -> ann.hasNonDefaultValue("value")).isPresent()
&& messagingAnnotations.size() > 1) {
throw new IllegalStateException("@EndpointId on " + identified
+ " can only have one EIP annotation, found: " + messagingAnnotations.size());
}
return messagingAnnotations;
}
private void processMessagingAnnotationOnBean(String beanName, AnnotatedBeanDefinition beanDefinition,
Class<? extends Annotation> annotationType, List<Annotation> annotationChain) {
@@ -220,150 +179,8 @@ public class MessagingAnnotationPostProcessor
this.postProcessors.put(annotation, postProcessor);
}
@Override
public void afterSingletonsInstantiated() {
this.initialized = true;
this.methodsToPostProcessAfterContextInitialization.forEach(Runnable::run);
this.methodsToPostProcessAfterContextInitialization.clear();
}
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
@Override
public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
Assert.notNull(this.beanFactory, "BeanFactory must not be null");
Class<?> beanClass = AopUtils.getTargetClass(bean);
// the set will hold records of prior class scans and indicate if no messaging annotations were found
if (this.noAnnotationsCache.contains(beanClass)) {
return bean;
}
ReflectionUtils.doWithMethods(beanClass,
method -> doWithMethod(method, bean, beanName, beanClass),
ReflectionUtils.USER_DECLARED_METHODS);
return bean;
}
private void doWithMethod(Method method, Object bean, String beanName, Class<?> beanClass) {
MergedAnnotations mergedAnnotations = MergedAnnotations.from(method);
boolean noMessagingAnnotations = true;
if (!mergedAnnotations.isPresent(Bean.class)) { // See postProcessBeanDefinitionRegistry(BeanDefinitionRegistry)
List<MessagingMetaAnnotation> messagingAnnotations =
obtainMessagingAnnotations(mergedAnnotations, method.toGenericString());
for (MessagingMetaAnnotation messagingAnnotation : messagingAnnotations) {
noMessagingAnnotations = false;
Class<? extends Annotation> annotationType = messagingAnnotation.annotationType;
List<Annotation> annotationChain =
MessagingAnnotationUtils.getAnnotationChain(messagingAnnotation.annotation, annotationType);
processAnnotationTypeOnMethod(bean, beanName, method, annotationType, annotationChain);
}
}
if (noMessagingAnnotations) {
this.noAnnotationsCache.add(beanClass);
}
}
protected void processAnnotationTypeOnMethod(Object bean, String beanName, Method method,
Class<? extends Annotation> annotationType, List<Annotation> annotations) {
MethodAnnotationPostProcessor<?> postProcessor =
MessagingAnnotationPostProcessor.this.postProcessors.get(annotationType);
if (postProcessor != null && postProcessor.supportsPojoMethod()
&& postProcessor.shouldCreateEndpoint(method, annotations)) {
Method targetMethod = method;
if (AopUtils.isJdkDynamicProxy(bean)) {
try {
targetMethod = bean.getClass().getMethod(method.getName(), method.getParameterTypes());
}
catch (NoSuchMethodException e) {
throw new IllegalArgumentException("Service methods must be extracted to the service "
+ "interface for JdkDynamicProxy. The affected bean is: '" + beanName + "' "
+ "and its method: '" + method + "'", e);
}
}
if (this.initialized) {
postProcessMethodAndRegisterEndpointIfAny(bean, beanName, method, annotationType, annotations,
postProcessor, targetMethod);
}
else {
Method methodToPostProcess = targetMethod;
this.methodsToPostProcessAfterContextInitialization.add(() ->
postProcessMethodAndRegisterEndpointIfAny(bean, beanName, method, annotationType, annotations,
postProcessor, methodToPostProcess));
}
}
}
@SuppressWarnings("unchecked")
private void postProcessMethodAndRegisterEndpointIfAny(Object bean, String beanName, Method method,
Class<? extends Annotation> annotationType, List<Annotation> annotations,
MethodAnnotationPostProcessor<?> postProcessor, Method targetMethod) {
Object result = postProcessor.postProcess(bean, beanName, targetMethod, annotations);
if (result instanceof AbstractEndpoint endpoint) {
String autoStartup = MessagingAnnotationUtils.resolveAttribute(annotations, "autoStartup", String.class);
if (StringUtils.hasText(autoStartup)) {
autoStartup = this.beanFactory.resolveEmbeddedValue(autoStartup);
if (StringUtils.hasText(autoStartup)) {
endpoint.setAutoStartup(Boolean.parseBoolean(autoStartup));
}
}
String phase = MessagingAnnotationUtils.resolveAttribute(annotations, "phase", String.class);
if (StringUtils.hasText(phase)) {
phase = this.beanFactory.resolveEmbeddedValue(phase);
if (StringUtils.hasText(phase)) {
endpoint.setPhase(Integer.parseInt(phase));
}
}
Role role = AnnotationUtils.findAnnotation(method, Role.class);
if (role != null) {
endpoint.setRole(role.value());
}
String endpointBeanName = generateBeanName(beanName, method, annotationType);
endpoint.setBeanName(endpointBeanName);
getBeanDefinitionRegistry()
.registerBeanDefinition(endpointBeanName,
new RootBeanDefinition((Class<AbstractEndpoint>) endpoint.getClass(), () -> endpoint));
this.beanFactory.getBean(endpointBeanName);
}
}
protected String generateBeanName(String originalBeanName, Method method,
Class<? extends Annotation> annotationType) {
String name = MessagingAnnotationUtils.endpointIdValue(method);
if (!StringUtils.hasText(name)) {
String baseName = originalBeanName + "." + method.getName() + "."
+ ClassUtils.getShortNameAsProperty(annotationType);
name = baseName;
int count = 1;
while (this.beanFactory.containsBean(name)) {
name = baseName + "#" + (++count);
}
}
return name;
}
protected Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> getPostProcessors() {
return this.postProcessors;
}
protected record MessagingMetaAnnotation(Annotation annotation, Class<? extends Annotation> annotationType) {
}
}

18
spring-integration-core/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -20,7 +20,6 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
@@ -55,6 +54,7 @@ public class DirectChannelSubscriptionTests {
@BeforeEach
public void setupChannels() {
this.context.registerBean(MessagingAnnotationPostProcessor.class);
this.context.registerChannel("sourceChannel", this.sourceChannel);
this.context.registerChannel("targetChannel", this.targetChannel);
}
@@ -80,13 +80,9 @@ public class DirectChannelSubscriptionTests {
@Test
public void sendAndReceiveForAnnotatedEndpoint() {
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
postProcessor.postProcessBeanDefinitionRegistry((BeanDefinitionRegistry) this.context.getBeanFactory());
postProcessor.postProcessBeanFactory(this.context.getBeanFactory());
postProcessor.afterSingletonsInstantiated();
TestEndpoint endpoint = new TestEndpoint();
postProcessor.postProcessAfterInitialization(endpoint, "testEndpoint");
this.context.registerEndpoint("testEndpoint", new TestEndpoint());
this.context.refresh();
this.sourceChannel.send(new GenericMessage<>("foo"));
Message<?> response = this.targetChannel.receive();
assertThat(response.getPayload()).isEqualTo("foo-from-annotated-endpoint");
@@ -113,11 +109,7 @@ public class DirectChannelSubscriptionTests {
public void exceptionThrownFromAnnotatedEndpoint() {
QueueChannel errorChannel = new QueueChannel();
this.context.registerChannel(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME, errorChannel);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
postProcessor.postProcessBeanFactory(this.context.getBeanFactory());
postProcessor.afterSingletonsInstantiated();
FailingTestEndpoint endpoint = new FailingTestEndpoint();
postProcessor.postProcessAfterInitialization(endpoint, "testEndpoint");
this.context.registerEndpoint("testEndpoint", new FailingTestEndpoint());
this.context.refresh();
assertThatExceptionOfType(MessagingException.class)
.isThrownBy(() -> this.sourceChannel.send(new GenericMessage<>("foo")));

25
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/FilterAnnotationPostProcessorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -23,11 +23,11 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.integration.annotation.Filter;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.MessagingAnnotationBeanPostProcessor;
import org.springframework.integration.config.MessagingAnnotationPostProcessor;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
@@ -50,19 +50,15 @@ public class FilterAnnotationPostProcessorTests {
private final TestApplicationContext context = TestUtils.createTestApplicationContext();
private final MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
private final DirectChannel inputChannel = new DirectChannel();
private final QueueChannel outputChannel = new QueueChannel();
@BeforeEach
public void init() {
this.context.registerBean(MessagingAnnotationPostProcessor.class);
this.context.registerChannel("input", this.inputChannel);
this.context.registerChannel("output", this.outputChannel);
this.postProcessor.postProcessBeanDefinitionRegistry((BeanDefinitionRegistry) this.context.getBeanFactory());
this.postProcessor.postProcessBeanFactory(this.context.getBeanFactory());
this.postProcessor.afterSingletonsInstantiated();
}
@AfterEach
@@ -175,21 +171,26 @@ public class FilterAnnotationPostProcessorTests {
@Test
public void invalidMethodWithStringReturnType() {
Object filter = new TestFilterWithStringReturnType();
context.refresh();
var postProcessor = context.getBean(MessagingAnnotationBeanPostProcessor.class);
assertThatIllegalArgumentException()
.isThrownBy(() -> this.postProcessor.postProcessAfterInitialization(filter, "testFilter"));
.isThrownBy(() ->
postProcessor.postProcessAfterInitialization(
new TestFilterWithStringReturnType(), "testFilter"));
}
@Test
public void invalidMethodWithVoidReturnType() {
Object filter = new TestFilterWithVoidReturnType();
context.refresh();
var postProcessor = context.getBean(MessagingAnnotationBeanPostProcessor.class);
assertThatIllegalArgumentException()
.isThrownBy(() -> postProcessor.postProcessAfterInitialization(filter, "testFilter"));
.isThrownBy(() ->
postProcessor.postProcessAfterInitialization(new TestFilterWithVoidReturnType(), "testFilter"));
}
private void testValidFilter(Object filter) {
postProcessor.postProcessAfterInitialization(filter, "testFilter");
context.registerEndpoint("testFilter", filter);
context.refresh();
inputChannel.send(new GenericMessage<>("good"));
Message<?> passed = outputChannel.receive(0);

56
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -27,8 +27,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.jupiter.api.Test;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.annotation.MessageEndpoint;
@@ -36,6 +34,7 @@ import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.MessagingAnnotationBeanPostProcessor;
import org.springframework.integration.config.MessagingAnnotationPostProcessor;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
@@ -63,9 +62,10 @@ public class MessagingAnnotationPostProcessorTests {
TestApplicationContext context = TestUtils.createTestApplicationContext();
DirectChannel inputChannel = new DirectChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerBean(MessagingAnnotationPostProcessor.class);
context.refresh();
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
MessagingAnnotationBeanPostProcessor postProcessor = context.getBean(MessagingAnnotationBeanPostProcessor.class);
ServiceActivatorAnnotatedBean bean = new ServiceActivatorAnnotatedBean();
postProcessor.postProcessAfterInitialization(bean, "testBean");
assertThat(context.containsBean("testBean.test.serviceActivator")).isTrue();
@@ -135,11 +135,11 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void outboundOnlyServiceActivator() throws InterruptedException {
TestApplicationContext context = TestUtils.createTestApplicationContext();
context.registerBean(MessagingAnnotationPostProcessor.class);
context.registerChannel("testChannel", new DirectChannel());
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
CountDownLatch latch = new CountDownLatch(1);
OutboundOnlyTestBean testBean = new OutboundOnlyTestBean(latch);
postProcessor.postProcessAfterInitialization(testBean, "testBean");
context.registerEndpoint("testBean", testBean);
context.refresh();
DestinationResolver<MessageChannel> channelResolver = new BeanFactoryChannelResolver(context);
MessageChannel testChannel = channelResolver.resolveDestination("testChannel");
@@ -153,15 +153,15 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testChannelResolution() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
context.registerBean(MessagingAnnotationPostProcessor.class);
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
DirectChannel eventBus = new DirectChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
context.registerChannel("eventBus", eventBus);
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
ServiceActivatorAnnotatedBean bean = new ServiceActivatorAnnotatedBean();
postProcessor.postProcessAfterInitialization(bean, "testBean");
context.registerEndpoint("testBean", bean);
context.refresh();
Message<?> message = MessageBuilder.withPayload("test")
.setReplyChannelName("outputChannel").build();
@@ -178,14 +178,14 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testProxiedMessageEndpointAnnotation() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
context.registerBean(MessagingAnnotationPostProcessor.class);
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
ProxyFactory proxyFactory = new ProxyFactory(new AnnotatedTestService());
Object proxy = proxyFactory.getProxy();
postProcessor.postProcessAfterInitialization(proxy, "proxy");
context.registerEndpoint("proxy", proxy);
context.refresh();
inputChannel.send(new GenericMessage<>("world"));
Message<?> message = outputChannel.receive(1000);
@@ -196,12 +196,12 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInherited() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
context.registerBean(MessagingAnnotationPostProcessor.class);
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointSubclass(), "subclass");
context.registerEndpoint("subclass", new SimpleAnnotatedEndpointSubclass());
context.refresh();
inputChannel.send(new GenericMessage<>("world"));
Message<?> message = outputChannel.receive(1000);
@@ -212,14 +212,14 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInheritedWithProxy() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
context.registerBean(MessagingAnnotationPostProcessor.class);
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointSubclass());
Object proxy = proxyFactory.getProxy();
postProcessor.postProcessAfterInitialization(proxy, "proxy");
context.registerEndpoint("proxy", proxy);
context.refresh();
inputChannel.send(new GenericMessage<>("world"));
Message<?> message = outputChannel.receive(1000);
@@ -230,12 +230,12 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInheritedFromInterface() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
context.registerBean(MessagingAnnotationPostProcessor.class);
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl");
context.registerEndpoint("impl", new SimpleAnnotatedEndpointImplementation());
context.refresh();
inputChannel.send(new GenericMessage<>("ABC"));
Message<?> message = outputChannel.receive(1000);
@@ -246,12 +246,12 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInheritedFromInterfaceWithAutoCreatedChannels() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
context.registerBean(MessagingAnnotationPostProcessor.class);
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl");
context.registerEndpoint("impl", new SimpleAnnotatedEndpointImplementation());
context.refresh();
inputChannel.send(new GenericMessage<>("ABC"));
Message<?> message = outputChannel.receive(1000);
@@ -262,14 +262,13 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInheritedFromInterfaceWithProxy() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
context.registerBean(MessagingAnnotationPostProcessor.class);
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointImplementation());
Object proxy = proxyFactory.getProxy();
postProcessor.postProcessAfterInitialization(proxy, "proxy");
context.registerEndpoint("proxy", proxyFactory.getProxy());
context.refresh();
inputChannel.send(new GenericMessage<>("ABC"));
Message<?> message = outputChannel.receive(1000);
@@ -280,13 +279,12 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testTransformer() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
context.registerBean(MessagingAnnotationPostProcessor.class);
DirectChannel inputChannel = new DirectChannel();
context.registerChannel("inputChannel", inputChannel);
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("outputChannel", outputChannel);
MessagingAnnotationPostProcessor postProcessor = prepareMessagingAnnotationPostProcessor(context);
TransformerAnnotationTestBean testBean = new TransformerAnnotationTestBean();
postProcessor.postProcessAfterInitialization(testBean, "testBean");
context.registerEndpoint("testBean", new TransformerAnnotationTestBean());
context.refresh();
inputChannel.send(new GenericMessage<>("foo"));
Message<?> reply = outputChannel.receive(0);
@@ -300,16 +298,6 @@ public class MessagingAnnotationPostProcessorTests {
context.close();
}
private static MessagingAnnotationPostProcessor prepareMessagingAnnotationPostProcessor(
ConfigurableApplicationContext context) {
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
postProcessor.postProcessBeanDefinitionRegistry((BeanDefinitionRegistry) context.getBeanFactory());
postProcessor.postProcessBeanFactory(context.getBeanFactory());
postProcessor.afterSingletonsInstantiated();
return postProcessor;
}
@MessageEndpoint
public static class OutboundOnlyTestBean {

15
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -23,7 +23,6 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.channel.DirectChannel;
@@ -44,8 +43,6 @@ public class RouterAnnotationPostProcessorTests {
private final TestApplicationContext context = TestUtils.createTestApplicationContext();
private final MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
private final DirectChannel inputChannel = new DirectChannel();
private final QueueChannel outputChannel = new QueueChannel();
@@ -59,14 +56,12 @@ public class RouterAnnotationPostProcessorTests {
@BeforeEach
public void init() {
this.context.registerBean(MessagingAnnotationPostProcessor.class);
context.registerChannel("input", inputChannel);
context.registerChannel("output", outputChannel);
context.registerChannel("routingChannel", routingChannel);
context.registerChannel("integerChannel", integerChannel);
context.registerChannel("stringChannel", stringChannel);
this.postProcessor.postProcessBeanDefinitionRegistry((BeanDefinitionRegistry) this.context.getBeanFactory());
this.postProcessor.postProcessBeanFactory(this.context.getBeanFactory());
this.postProcessor.afterSingletonsInstantiated();
}
@AfterEach
@@ -76,8 +71,7 @@ public class RouterAnnotationPostProcessorTests {
@Test
public void testRouter() {
TestRouter testRouter = new TestRouter();
postProcessor.postProcessAfterInitialization(testRouter, "test");
context.registerEndpoint("test", new TestRouter());
context.refresh();
inputChannel.send(new GenericMessage<>("foo"));
Message<?> replyMessage = outputChannel.receive(0);
@@ -87,8 +81,7 @@ public class RouterAnnotationPostProcessorTests {
@Test
public void testRouterWithListParam() {
TestRouter testRouter = new TestRouter();
postProcessor.postProcessAfterInitialization(testRouter, "test");
context.registerEndpoint("test", new TestRouter());
context.refresh();
routingChannel.send(new GenericMessage<>(Collections.singletonList("foo")));

10
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -20,7 +20,6 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.context.Lifecycle;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Splitter;
@@ -50,6 +49,7 @@ public class SplitterAnnotationPostProcessorTests {
@BeforeEach
public void init() {
this.context.registerBean(MessagingAnnotationPostProcessor.class);
this.context.registerChannel("input", this.inputChannel);
this.context.registerChannel("output", this.outputChannel);
}
@@ -61,12 +61,8 @@ public class SplitterAnnotationPostProcessorTests {
@Test
public void testSplitterAnnotation() {
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
postProcessor.postProcessBeanDefinitionRegistry((BeanDefinitionRegistry) this.context.getBeanFactory());
postProcessor.postProcessBeanFactory(this.context.getBeanFactory());
postProcessor.afterSingletonsInstantiated();
TestSplitter splitter = new TestSplitter();
postProcessor.postProcessAfterInitialization(splitter, "testSplitter");
context.registerEndpoint("testSplitter", splitter);
context.refresh();
inputChannel.send(new GenericMessage<>("this.is.a.test"));
Message<?> message1 = outputChannel.receive(500);