From 80fee48b4a51ef6123086b2b6920b6e1b628c0a5 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 --- ...ListenerAnnotatedMethodArgumentsTests.java | 5 +- .../StreamListenerHandlerMethodTests.java | 52 ++--- ...enerWithAnnotatedInputOutputArgsTests.java | 9 +- .../StreamListenerWithConditionsTest.java | 12 +- ...amListenerAnnotationBeanPostProcessor.java | 219 +++++++++--------- 5 files changed, 148 insertions(+), 149 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 a200bdc1d..d8b53c62a 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; @@ -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); } } 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 811bb0353..c34f86d47 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 @@ -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); } } 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 f704ca866..88bcca9cc 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; @@ -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); } } 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 a77991e8a..d47efd038 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; @@ -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); } } 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 3dc661fe6..6839a4337 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,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 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; @@ -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> 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(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> 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(); } 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;