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

polished for Java 1.7 compliance
This commit is contained in:
Oleg Zhurakousky
2017-12-13 11:07:25 -05:00
parent 85ede8a3fd
commit 11ea4b3436
5 changed files with 162 additions and 153 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;
@@ -80,8 +79,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

@@ -22,7 +22,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;
@@ -73,8 +72,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestInvalidInputOnMethod.class, "--server.port=0");
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);
}
}
@@ -149,8 +148,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestReturnTypeWithMultipleOutput.class, "--server.port=0");
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);
}
}
@@ -160,8 +159,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestInvalidReturnTypeWithNoOutput.class, "--server.port=0");
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);
}
}
@@ -171,8 +170,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestInvalidInputAnnotationWithNoValue.class, "--server.port=0");
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);
}
}
@@ -182,8 +181,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestInvalidOutputAnnotationWithNoValue.class, "--server.port=0");
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);
}
}
@@ -193,10 +192,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestMethodInvalidInboundName.class, "--server.port=0");
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);
}
}
@@ -206,9 +203,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestMethodInvalidOutboundName.class, "--server.port=0");
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");
}
}
@@ -218,8 +214,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestAmbiguousMethodArguments1.class, "--server.port=0");
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);
}
}
@@ -229,8 +225,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestAmbiguousMethodArguments2.class, "--server.port=0");
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);
}
}
@@ -240,8 +236,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestMethodWithInputAsMethodAndParameter.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);
}
}
@@ -251,8 +247,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestMethodWithOutputAsMethodAndParameter.class, "--server.port=0");
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);
}
}
@@ -262,8 +258,8 @@ public class StreamListenerHandlerMethodTests {
SpringApplication.run(TestMethodWithoutInput.class, "--server.port=0");
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;
@@ -60,7 +59,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);
}
}
@@ -71,10 +70,8 @@ public class StreamListenerWithAnnotatedInputOutputArgsTests {
SpringApplication.run(TestInputOutputArgsWithInvalidBindableTarget.class, "--server.port=0");
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;
@@ -81,10 +80,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);
}
}
@@ -98,9 +95,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);
}
}

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

@@ -18,15 +18,18 @@ 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.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.springframework.aop.framework.Advised;
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;
@@ -37,7 +40,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;
@@ -57,8 +59,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.
@@ -67,31 +67,27 @@ import org.springframework.util.StringUtils;
* @author Ilayaperumal Gopinathan
* @author Soby Chacko
*/
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;
@@ -114,38 +110,97 @@ public class StreamListenerAnnotationBeanPostProcessor
}
@Override
public final Object postProcessAfterInitialization(Object bean, final String beanName) throws BeansException {
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(final Object bean, final String beanName) throws BeansException {
Class<?> targetClass = AopUtils.isAopProxy(bean) ? AopUtils.getTargetClass(bean) : bean.getClass();
Method[] uniqueDeclaredMethods = ReflectionUtils.getUniqueDeclaredMethods(targetClass);
for (Method method : uniqueDeclaredMethods) {
StreamListener streamListener = AnnotatedElementUtils.findMergedAnnotation(method, StreamListener.class);
for (final Method method : uniqueDeclaredMethods) {
final 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(new Runnable() {
@Override
public void run() {
Assert.isTrue(method.getAnnotation(Input.class) == null, StreamListenerErrorMessages.INPUT_AT_STREAM_LISTENER);
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()));
}
/**
@@ -202,14 +257,17 @@ public class StreamListenerAnnotationBeanPostProcessor
* there is a {@link StreamListenerParameterAdapter} (i.e., {@link Flux}). Declarative method is invoked only
* once during initialization phase.
*/
@SuppressWarnings("rawtypes")
private boolean isDeclarativeMethodParameter(String targetBeanName, MethodParameter methodParameter) {
boolean declarative = false;
if (!methodParameter.getParameterType().isAssignableFrom(Object.class) && this.applicationContext.containsBean(targetBeanName)) {
declarative = MessageChannel.class.isAssignableFrom(methodParameter.getParameterType());
if (!declarative) {
Class<?> targetBeanClass = this.applicationContext.getType(targetBeanName);
declarative = this.streamListenerParameterAdapters.stream()
.filter(slpa -> slpa.supports(targetBeanClass, methodParameter)).findFirst().isPresent();
Iterator<StreamListenerParameterAdapter> slpaIter = this.streamListenerParameterAdapters.iterator();
while (slpaIter.hasNext() && !declarative) {
declarative = slpaIter.next().supports(targetBeanClass, methodParameter);
}
}
}
return declarative;
@@ -280,77 +338,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();
applicationContext.getBean(mappedBindingEntry.getKey(), SubscribableChannel.class).subscribe(handler);
}
this.mappedListenerMethods.clear();
}
private Method checkProxy(Method methodArg, Object bean) {
@@ -422,6 +428,21 @@ 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);
for (Runnable runnable : streamListenerCallbacks) {
runnable.run();
}
//this.streamListenerCallbacks.forEach(r -> r.run());
}
private class StreamListenerHandlerMethodMapping {
private final Object targetBean;