SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

1146 Fixed early initialization in StreamListenerAnnotationBeanPostProcessor

Browse Source 
Removed autowiring from StreamListenerAnnotationBeanPostProcessor in favor of late-binding callbacks
Fixed tests

Resolves #1146
This commit is contained in:
Oleg Zhurakousky
2017-12-13 11:07:25 -05:00
parent f59c85842b
commit 80fee48b4a
5 changed files with 148 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java
View File

@@ -23,7 +23,6 @@ import java.util.UUID;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
@@ -82,8 +81,8 @@ public class StreamListenerAnnotatedMethodArgumentsTests {
SpringApplication.run(TestPojoWithInvalidInputAnnotatedArgument.class, "--server.port=0");
fail("Exception expected: " + INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INVALID_DECLARATIVE_METHOD_PARAMETERS);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}

52
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java
View File

@@ -21,7 +21,6 @@ import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -75,8 +74,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected: " + INPUT_AT_STREAM_LISTENER);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INPUT_AT_STREAM_LISTENER);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(INPUT_AT_STREAM_LISTENER);
}
}
@@ -174,8 +173,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected: " + RETURN_TYPE_MULTIPLE_OUTBOUND_SPECIFIED);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(RETURN_TYPE_MULTIPLE_OUTBOUND_SPECIFIED);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(RETURN_TYPE_MULTIPLE_OUTBOUND_SPECIFIED);
}
}
@@ -186,8 +185,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected: " + RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(RETURN_TYPE_NO_OUTBOUND_SPECIFIED);
}
}
@@ -198,8 +197,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected: " + INVALID_INBOUND_NAME);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INVALID_INBOUND_NAME);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(INVALID_INBOUND_NAME);
}
}
@@ -210,8 +209,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected: " + INVALID_OUTBOUND_NAME);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INVALID_OUTBOUND_NAME);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(INVALID_OUTBOUND_NAME);
}
}
@@ -222,10 +221,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected on using invalid inbound name");
}
catch (BeanCreationException e) {
assertThat(e.getCause()).isInstanceOf(IllegalArgumentException.class);
assertThat(e.getCause())
.hasMessageContaining(StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}
@@ -236,9 +233,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected on using invalid outbound name");
}
catch (BeanCreationException e) {
assertThat(e.getCause()).isInstanceOf(NoSuchBeanDefinitionException.class);
assertThat(e.getCause()).hasMessageContaining("'invalid'");
catch (NoSuchBeanDefinitionException e) {
assertThat(e.getMessage()).contains("invalid");
}
}
@@ -249,8 +245,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected: " + AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
}
}
@@ -261,8 +257,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected:" + AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(AMBIGUOUS_MESSAGE_HANDLER_METHOD_ARGUMENTS);
}
}
@@ -273,8 +269,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected: " + INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(INVALID_DECLARATIVE_METHOD_PARAMETERS);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}
@@ -285,8 +281,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected:" + INVALID_OUTPUT_VALUES);
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).startsWith(INVALID_OUTPUT_VALUES);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).startsWith(INVALID_OUTPUT_VALUES);
}
}
@@ -297,8 +293,8 @@ public class StreamListenerHandlerMethodTests {
"--spring.jmx.enabled=false");
fail("Exception expected when inbound target is not set");
}
catch (BeanCreationException e) {
assertThat(e.getCause().getMessage()).contains(NO_INPUT_DESTINATION);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(NO_INPUT_DESTINATION);
}
}

9
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithAnnotatedInputOutputArgsTests.java
View File

@@ -20,7 +20,6 @@ import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
@@ -62,7 +61,7 @@ public class StreamListenerWithAnnotatedInputOutputArgsTests {
SpringApplication.run(TestInputOutputArgsWithMoreParameters.class, "--server.port=0");
fail("Expected exception: " + INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
catch (BeanCreationException e) {
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}
@@ -73,10 +72,8 @@ public class StreamListenerWithAnnotatedInputOutputArgsTests {
SpringApplication.run(TestInputOutputArgsWithInvalidBindableTarget.class, "--server.port=0","--spring.jmx.enabled=false");
fail("Exception expected on using invalid bindable target as method parameter");
}
catch (BeanCreationException e) {
assertThat(e.getCause()).isInstanceOf(IllegalArgumentException.class);
assertThat(e.getCause())
.hasMessageContaining(StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS);
}
}

12
spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java
View File

@@ -22,7 +22,6 @@ import java.util.UUID;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
@@ -79,10 +78,8 @@ public class StreamListenerWithConditionsTest {
context.close();
fail("Context creation failure expected");
}
catch (BeanCreationException e) {
assertThat(e).hasRootCauseInstanceOf(IllegalArgumentException.class);
assertThat(e.getCause())
.hasMessageContaining(StreamListenerErrorMessages.CONDITION_ON_METHOD_RETURNING_VALUE);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(StreamListenerErrorMessages.CONDITION_ON_METHOD_RETURNING_VALUE);
}
}
@@ -95,9 +92,8 @@ public class StreamListenerWithConditionsTest {
context.close();
fail("Context creation failure expected");
}
catch (BeanCreationException e) {
assertThat(e).hasRootCauseInstanceOf(IllegalArgumentException.class);
assertThat(e.getCause()).hasMessageContaining(StreamListenerErrorMessages.CONDITION_ON_DECLARATIVE_METHOD);
catch (IllegalArgumentException e) {
assertThat(e.getMessage()).contains(StreamListenerErrorMessages.CONDITION_ON_DECLARATIVE_METHOD);
}
}

219
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java
View File

@@ -18,8 +18,11 @@ package org.springframework.cloud.stream.binding;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import reactor.core.publisher.Flux;
@@ -28,7 +31,6 @@ import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanExpressionContext;
import org.springframework.beans.factory.config.BeanExpressionResolver;
import org.springframework.beans.factory.config.BeanPostProcessor;
@@ -39,7 +41,6 @@ import org.springframework.cloud.stream.config.SpringIntegrationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.MethodParameter;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.core.annotation.AnnotationUtils;
@@ -59,8 +60,6 @@ import org.springframework.util.MultiValueMap;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* {@link BeanPostProcessor} that handles {@link StreamListener} annotations found on bean
* methods.
@@ -70,31 +69,27 @@ import org.springframework.util.StringUtils;
* @author Soby Chacko
* @author Oleg Zhurakousky
*/
public class StreamListenerAnnotationBeanPostProcessor
implements BeanPostProcessor, ApplicationContextAware, SmartInitializingSingleton {
public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProcessor, ApplicationContextAware, SmartInitializingSingleton {
private static final SpelExpressionParser SPEL_EXPRESSION_PARSER = new SpelExpressionParser();
private final MultiValueMap<String, StreamListenerHandlerMethodMapping> mappedListenerMethods = new LinkedMultiValueMap<>();
@Autowired(required=false)
@Lazy
private List<StreamListenerParameterAdapter<?,?>> streamListenerParameterAdapters;
// == dependencies that are injected in 'afterSingletonsInstantiated' to avoid early initialization
@SuppressWarnings("rawtypes")
private Collection<StreamListenerParameterAdapter> streamListenerParameterAdapters;
@Autowired(required=false)
@Lazy
private List<StreamListenerResultAdapter<?, ?>> streamListenerResultAdapters;
@SuppressWarnings("rawtypes")
private Collection<StreamListenerResultAdapter> streamListenerResultAdapters;
@Autowired
@Lazy
private DestinationResolver<MessageChannel> binderAwareChannelResolver;
@Autowired
@Lazy
private MessageHandlerMethodFactory messageHandlerMethodFactory;
@Autowired
private SpringIntegrationProperties springIntegrationProperties;
// == end dependencies
private final Set<Runnable> streamListenerCallbacks = new HashSet<>();
private ConfigurableApplicationContext applicationContext;
@@ -111,6 +106,61 @@ public class StreamListenerAnnotationBeanPostProcessor
this.expressionContext = new BeanExpressionContext(this.applicationContext.getBeanFactory(), null);
}
@Override
public final void afterSingletonsInstantiated() {
this.injectAndPostProcessDependencies();
this.evaluationContext = IntegrationContextUtils.getEvaluationContext(this.applicationContext.getBeanFactory());
for (Map.Entry<String, List<StreamListenerHandlerMethodMapping>> mappedBindingEntry : mappedListenerMethods
.entrySet()) {
ArrayList<DispatchingStreamListenerMessageHandler.ConditionalStreamListenerMessageHandlerWrapper> handlers = new ArrayList<>();
for (StreamListenerHandlerMethodMapping mapping : mappedBindingEntry.getValue()) {
final InvocableHandlerMethod invocableHandlerMethod = this.messageHandlerMethodFactory
.createInvocableHandlerMethod(mapping.getTargetBean(),
checkProxy(mapping.getMethod(), mapping.getTargetBean()));
StreamListenerMessageHandler streamListenerMessageHandler = new StreamListenerMessageHandler(
invocableHandlerMethod, resolveExpressionAsBoolean(mapping.getCopyHeaders(), "copyHeaders"),
springIntegrationProperties.getMessageHandlerNotPropagatedHeaders());
streamListenerMessageHandler.setApplicationContext(this.applicationContext);
streamListenerMessageHandler.setBeanFactory(this.applicationContext.getBeanFactory());
if (StringUtils.hasText(mapping.getDefaultOutputChannel())) {
streamListenerMessageHandler.setOutputChannelName(mapping.getDefaultOutputChannel());
}
streamListenerMessageHandler.afterPropertiesSet();
if (StringUtils.hasText(mapping.getCondition())) {
String conditionAsString = resolveExpressionAsString(mapping.getCondition(), "condition");
Expression condition = SPEL_EXPRESSION_PARSER.parseExpression(conditionAsString);
handlers.add(
new DispatchingStreamListenerMessageHandler.ConditionalStreamListenerMessageHandlerWrapper(
condition, streamListenerMessageHandler));
}
else {
handlers.add(
new DispatchingStreamListenerMessageHandler.ConditionalStreamListenerMessageHandlerWrapper(
null, streamListenerMessageHandler));
}
}
if (handlers.size() > 1) {
for (DispatchingStreamListenerMessageHandler.ConditionalStreamListenerMessageHandlerWrapper handler : handlers) {
Assert.isTrue(handler.isVoid(), StreamListenerErrorMessages.MULTIPLE_VALUE_RETURNING_METHODS);
}
}
AbstractReplyProducingMessageHandler handler;
if (handlers.size() > 1 || handlers.get(0).getCondition() != null) {
handler = new DispatchingStreamListenerMessageHandler(handlers, this.evaluationContext);
}
else {
handler = handlers.get(0).getStreamListenerMessageHandler();
}
handler.setApplicationContext(this.applicationContext);
handler.setChannelResolver(this.binderAwareChannelResolver);
handler.afterPropertiesSet();
this.applicationContext.getBeanFactory().registerSingleton(handler.getClass().getSimpleName() + handler.hashCode(), handler);
applicationContext.getBean(mappedBindingEntry.getKey(), SubscribableChannel.class).subscribe(handler);
}
this.mappedListenerMethods.clear();
}
@Override
public final Object postProcessAfterInitialization(Object bean, final String beanName) throws BeansException {
Class<?> targetClass = AopUtils.isAopProxy(bean) ? AopUtils.getTargetClass(bean) : bean.getClass();
@@ -118,32 +168,33 @@ public class StreamListenerAnnotationBeanPostProcessor
for (Method method : uniqueDeclaredMethods) {
StreamListener streamListener = AnnotatedElementUtils.findMergedAnnotation(method, StreamListener.class);
if (streamListener != null && !method.isBridge()) {
Assert.isTrue(method.getAnnotation(Input.class) == null, StreamListenerErrorMessages.INPUT_AT_STREAM_LISTENER);
this.doPostProcess(streamListener, method, bean);
streamListenerCallbacks.add(() -> {
Assert.isTrue(method.getAnnotation(Input.class) == null, StreamListenerErrorMessages.INPUT_AT_STREAM_LISTENER);
this.doPostProcess(streamListener, method, bean);
});
}
}
return bean;
}
private void doPostProcess(StreamListener streamListener, Method method, Object bean) {
streamListener = postProcessAnnotation(streamListener, method);
String methodAnnotatedInboundName = streamListener.value();
String methodAnnotatedOutboundName = StreamListenerMethodUtils.getOutboundBindingTargetName(method);
int inputAnnotationCount = StreamListenerMethodUtils.inputAnnotationCount(method);
int outputAnnotationCount = StreamListenerMethodUtils.outputAnnotationCount(method);
boolean isDeclarative = checkDeclarativeMethod(method, methodAnnotatedInboundName, methodAnnotatedOutboundName);
StreamListenerMethodUtils.validateStreamListenerMethod(method,
inputAnnotationCount, outputAnnotationCount,
methodAnnotatedInboundName, methodAnnotatedOutboundName,
isDeclarative, streamListener.condition());
if (isDeclarative) {
invokeSetupMethodOnListenedChannel(method, bean, methodAnnotatedInboundName, methodAnnotatedOutboundName);
protected final void registerHandlerMethodOnListenedChannel(Method method, StreamListener streamListener, Object bean) {
Assert.hasText(streamListener.value(), "The binding name cannot be null");
if (!StringUtils.hasText(streamListener.value())) {
throw new BeanInitializationException("A bound component name must be specified");
}
final String defaultOutputChannel = StreamListenerMethodUtils.getOutboundBindingTargetName(method);
if (Void.TYPE.equals(method.getReturnType())) {
Assert.isTrue(StringUtils.isEmpty(defaultOutputChannel),
"An output channel cannot be specified for a method that does not return a value");
}
else {
registerHandlerMethodOnListenedChannel(method, streamListener, bean);
Assert.isTrue(!StringUtils.isEmpty(defaultOutputChannel),
"An output channel must be specified for a method that can return a value");
}
StreamListenerMethodUtils.validateStreamListenerMessageHandler(method);
mappedListenerMethods.add(streamListener.value(),
new StreamListenerHandlerMethodMapping(bean, method, streamListener.condition(), defaultOutputChannel,
streamListener.copyHeaders()));
}
/**
@@ -200,6 +251,7 @@ public class StreamListenerAnnotationBeanPostProcessor
* there is a {@link StreamListenerParameterAdapter} (i.e., {@link Flux}). Declarative method is invoked only
* once during initialization phase.
*/
@SuppressWarnings("unchecked")
private boolean isDeclarativeMethodParameter(String targetBeanName, MethodParameter methodParameter) {
boolean declarative = false;
if (!methodParameter.getParameterType().isAssignableFrom(Object.class) && this.applicationContext.containsBean(targetBeanName)) {
@@ -277,78 +329,25 @@ public class StreamListenerAnnotationBeanPostProcessor
}
}
protected final void registerHandlerMethodOnListenedChannel(Method method, StreamListener streamListener, Object bean) {
Assert.hasText(streamListener.value(), "The binding name cannot be null");
if (!StringUtils.hasText(streamListener.value())) {
throw new BeanInitializationException("A bound component name must be specified");
}
final String defaultOutputChannel = StreamListenerMethodUtils.getOutboundBindingTargetName(method);
if (Void.TYPE.equals(method.getReturnType())) {
Assert.isTrue(StringUtils.isEmpty(defaultOutputChannel),
"An output channel cannot be specified for a method that does not return a value");
private void doPostProcess(StreamListener streamListener, Method method, Object bean) {
streamListener = postProcessAnnotation(streamListener, method);
String methodAnnotatedInboundName = streamListener.value();
String methodAnnotatedOutboundName = StreamListenerMethodUtils.getOutboundBindingTargetName(method);
int inputAnnotationCount = StreamListenerMethodUtils.inputAnnotationCount(method);
int outputAnnotationCount = StreamListenerMethodUtils.outputAnnotationCount(method);
boolean isDeclarative = checkDeclarativeMethod(method, methodAnnotatedInboundName, methodAnnotatedOutboundName);
StreamListenerMethodUtils.validateStreamListenerMethod(method,
inputAnnotationCount, outputAnnotationCount,
methodAnnotatedInboundName, methodAnnotatedOutboundName,
isDeclarative, streamListener.condition());
if (isDeclarative) {
invokeSetupMethodOnListenedChannel(method, bean, methodAnnotatedInboundName, methodAnnotatedOutboundName);
}
else {
Assert.isTrue(!StringUtils.isEmpty(defaultOutputChannel),
"An output channel must be specified for a method that can return a value");
registerHandlerMethodOnListenedChannel(method, streamListener, bean);
}
StreamListenerMethodUtils.validateStreamListenerMessageHandler(method);
mappedListenerMethods.add(streamListener.value(),
new StreamListenerHandlerMethodMapping(bean, method, streamListener.condition(), defaultOutputChannel,
streamListener.copyHeaders()));
}
@Override
public final void afterSingletonsInstantiated() {
this.evaluationContext = IntegrationContextUtils.getEvaluationContext(this.applicationContext.getBeanFactory());
for (Map.Entry<String, List<StreamListenerHandlerMethodMapping>> mappedBindingEntry : mappedListenerMethods
.entrySet()) {
ArrayList<DispatchingStreamListenerMessageHandler.ConditionalStreamListenerMessageHandlerWrapper> handlers = new ArrayList<>();
for (StreamListenerHandlerMethodMapping mapping : mappedBindingEntry.getValue()) {
final InvocableHandlerMethod invocableHandlerMethod = this.messageHandlerMethodFactory
.createInvocableHandlerMethod(mapping.getTargetBean(),
checkProxy(mapping.getMethod(), mapping.getTargetBean()));
StreamListenerMessageHandler streamListenerMessageHandler = new StreamListenerMessageHandler(
invocableHandlerMethod, resolveExpressionAsBoolean(mapping.getCopyHeaders(), "copyHeaders"),
springIntegrationProperties.getMessageHandlerNotPropagatedHeaders());
streamListenerMessageHandler.setApplicationContext(this.applicationContext);
streamListenerMessageHandler.setBeanFactory(this.applicationContext.getBeanFactory());
if (StringUtils.hasText(mapping.getDefaultOutputChannel())) {
streamListenerMessageHandler.setOutputChannelName(mapping.getDefaultOutputChannel());
}
streamListenerMessageHandler.afterPropertiesSet();
if (StringUtils.hasText(mapping.getCondition())) {
String conditionAsString = resolveExpressionAsString(mapping.getCondition(), "condition");
Expression condition = SPEL_EXPRESSION_PARSER.parseExpression(conditionAsString);
handlers.add(
new DispatchingStreamListenerMessageHandler.ConditionalStreamListenerMessageHandlerWrapper(
condition, streamListenerMessageHandler));
}
else {
handlers.add(
new DispatchingStreamListenerMessageHandler.ConditionalStreamListenerMessageHandlerWrapper(
null, streamListenerMessageHandler));
}
}
if (handlers.size() > 1) {
for (DispatchingStreamListenerMessageHandler.ConditionalStreamListenerMessageHandlerWrapper handler : handlers) {
Assert.isTrue(handler.isVoid(), StreamListenerErrorMessages.MULTIPLE_VALUE_RETURNING_METHODS);
}
}
AbstractReplyProducingMessageHandler handler;
if (handlers.size() > 1 || handlers.get(0).getCondition() != null) {
handler = new DispatchingStreamListenerMessageHandler(handlers, this.evaluationContext);
}
else {
handler = handlers.get(0).getStreamListenerMessageHandler();
}
handler.setApplicationContext(this.applicationContext);
handler.setChannelResolver(this.binderAwareChannelResolver);
handler.afterPropertiesSet();
this.applicationContext.getBeanFactory().registerSingleton(handler.getClass().getSimpleName() + handler.hashCode(), handler);
applicationContext.getBean(mappedBindingEntry.getKey(), SubscribableChannel.class).subscribe(handler);
}
this.mappedListenerMethods.clear();
}
private Method checkProxy(Method methodArg, Object bean) {
@@ -420,6 +419,18 @@ public class StreamListenerAnnotationBeanPostProcessor
return resolvedValue;
}
/**
* This operations ensures that required dependencies are not accidently injected early given that this bean is BPP.
*/
private void injectAndPostProcessDependencies() {
this.streamListenerParameterAdapters = this.applicationContext.getBeansOfType(StreamListenerParameterAdapter.class).values();
this.streamListenerResultAdapters = this.applicationContext.getBeansOfType(StreamListenerResultAdapter.class).values();
this.binderAwareChannelResolver = this.applicationContext.getBean(DestinationResolver.class);
this.messageHandlerMethodFactory = this.applicationContext.getBean(MessageHandlerMethodFactory.class);
this.springIntegrationProperties = this.applicationContext.getBean(SpringIntegrationProperties.class);
this.streamListenerCallbacks.forEach(r -> r.run());
}
private class StreamListenerHandlerMethodMapping {
private final Object targetBean;