From 11ea4b3436bca1f453fb611c2fb113c5e129ed9e Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 13 Dec 2017 11:07:25 -0500 Subject: [PATCH] 1146 Fixed early initialization in StreamListenerAnnotationBeanPostProcessor Removed autowiring from StreamListenerAnnotationBeanPostProcessor in favor of late-binding callbacks Fixed tests Resolves #1146 polished for Java 1.7 compliance --- ...ListenerAnnotatedMethodArgumentsTests.java | 5 +- .../StreamListenerHandlerMethodTests.java | 52 ++-- ...enerWithAnnotatedInputOutputArgsTests.java | 9 +- .../StreamListenerWithConditionsTest.java | 12 +- ...amListenerAnnotationBeanPostProcessor.java | 237 ++++++++++-------- 5 files changed, 162 insertions(+), 153 deletions(-) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java index cc4ec8cf9..10fe8464a 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotatedMethodArgumentsTests.java @@ -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); } } diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java index abe449fc2..31da0e05e 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerHandlerMethodTests.java @@ -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); } } diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithAnnotatedInputOutputArgsTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithAnnotatedInputOutputArgsTests.java index 69b9e9621..707a572e3 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithAnnotatedInputOutputArgsTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithAnnotatedInputOutputArgsTests.java @@ -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); } } diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java index 73cc0a261..7d3a14378 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java @@ -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); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java index 8622cfe86..3d242d176 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java @@ -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 mappedListenerMethods = new LinkedMultiValueMap<>(); - @Autowired(required=false) - @Lazy - private List> streamListenerParameterAdapters; + // == dependencies that are injected in 'afterSingletonsInstantiated' to avoid early initialization + @SuppressWarnings("rawtypes") + private Collection streamListenerParameterAdapters; - @Autowired(required=false) - @Lazy - private List> streamListenerResultAdapters; + @SuppressWarnings("rawtypes") + private Collection streamListenerResultAdapters; - @Autowired - @Lazy private DestinationResolver binderAwareChannelResolver; - @Autowired - @Lazy private MessageHandlerMethodFactory messageHandlerMethodFactory; - @Autowired private SpringIntegrationProperties springIntegrationProperties; + // == end dependencies + + private final Set 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> mappedBindingEntry : mappedListenerMethods + .entrySet()) { + ArrayList 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 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> mappedBindingEntry : mappedListenerMethods - .entrySet()) { - ArrayList 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;