From 076f0ac1cb81f717cef70ad2a853b506560e4c5d Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Wed, 25 Jan 2017 18:16:08 -0500 Subject: [PATCH] Add Dispatching Capabilities to `@StreamListener` `@StreamListener` has support for a `condition` parameter, that contains a SpEL expression that is evaluated before the method is invoked. Fix #682 Move StreamListenerMessageHandler as a top level class Use dispatching and add test Refactor dispatching mechanism Throw error when conditions are used in declarative mode Make StreamListenerAnnotationBeanPostProcessor overridable Remove unused field in test Address some PR comments Add placeholder resolution Update how multiple matches work with return values - Methods with return values are not allowed to specify conditions - If multiple matches are detected (e.g. multiple methods without conditions, or a mix of methods with and without conditions) checks that all of them have no return value; --- ...notationBeanPostProcessorOverrideTest.java | 97 +++++++ .../StreamListenerDuplicateMappingTests.java | 27 +- .../StreamListenerWithConditionsTest.java | 141 ++++++++++ .../stream/annotation/StreamListener.java | 17 +- ...spatchingStreamListenerMessageHandler.java | 125 +++++++++ ...amListenerAnnotationBeanPostProcessor.java | 243 +++++++++++++----- .../binding/StreamListenerErrorMessages.java | 6 + .../binding/StreamListenerMessageHandler.java | 60 +++++ .../binding/StreamListenerMethodUtils.java | 10 +- .../config/BindingServiceConfiguration.java | 16 +- 10 files changed, 649 insertions(+), 93 deletions(-) create mode 100644 spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotationBeanPostProcessorOverrideTest.java create mode 100644 spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DispatchingStreamListenerMessageHandler.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerMessageHandler.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotationBeanPostProcessorOverrideTest.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotationBeanPostProcessorOverrideTest.java new file mode 100644 index 000000000..016ab779a --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerAnnotationBeanPostProcessorOverrideTest.java @@ -0,0 +1,97 @@ +/* + * Copyright 2016-2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.config; + +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import org.junit.Test; + +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.handler.annotation.Payload; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.cloud.stream.config.BindingServiceConfiguration.STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME; + +/** + * @author Marius Bogoevici + */ +public class StreamListenerAnnotationBeanPostProcessorOverrideTest { + + @Test + @SuppressWarnings("unchecked") + public void testOverrideStreamListenerAnnotationBeanPostProcessor() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithAnnotatedArguments.class, + "--server.port=0"); + + TestPojoWithAnnotatedArguments testPojoWithAnnotatedArguments = context + .getBean(TestPojoWithAnnotatedArguments.class); + Sink sink = context.getBean(Sink.class); + String id = UUID.randomUUID().toString(); + sink.input().send(MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").setHeader("testHeader", "testValue") + .setHeader("type", "foo").build()); + sink.input().send(MessageBuilder.withPayload("{\"bar\":\"foofoo" + id + "\"}") + .setHeader("contentType", "application/json").setHeader("testHeader", "testValue") + .setHeader("type", "bar").build()); + assertThat(testPojoWithAnnotatedArguments.receivedFoo).hasSize(1); + assertThat(testPojoWithAnnotatedArguments.receivedFoo.get(0)).hasFieldOrPropertyWithValue("foo", + "barbar" + id); + context.close(); + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + public static class TestPojoWithAnnotatedArguments { + + List receivedFoo = new ArrayList<>(); + + @StreamListener(value = Sink.INPUT, condition = "foo") + public void receive(@Payload StreamListenerTestUtils.FooPojo fooPojo) { + this.receivedFoo.add(fooPojo); + } + + /** + * Overrides the default {@link StreamListenerAnnotationBeanPostProcessor}. + */ + @Bean(name = STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME) + public static BeanPostProcessor streamListenerAnnotationBeanPostProcessor() { + return new StreamListenerAnnotationBeanPostProcessor() { + @Override + protected StreamListener postProcessAnnotation(StreamListener originalAnnotation, Method annotatedMethod) { + Map attributes = new HashMap<>(AnnotationUtils.getAnnotationAttributes(originalAnnotation)); + attributes.put("condition", "headers['type']=='" + originalAnnotation.condition() + "'"); + return AnnotationUtils.synthesizeAnnotation(attributes, StreamListener.class, annotatedMethod); + } + }; + } + } +} diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerDuplicateMappingTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerDuplicateMappingTests.java index c440aa7df..d482c48fe 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerDuplicateMappingTests.java +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerDuplicateMappingTests.java @@ -23,9 +23,12 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binding.StreamListenerErrorMessages; +import org.springframework.cloud.stream.messaging.Processor; import org.springframework.cloud.stream.messaging.Sink; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.messaging.Message; +import org.springframework.messaging.handler.annotation.SendTo; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.fail; @@ -38,14 +41,14 @@ public class StreamListenerDuplicateMappingTests { @Test @SuppressWarnings("unchecked") - public void testDuplicateMapping() { + public void testMultipleMappingsWithReturnValue() { ConfigurableApplicationContext context = null; try { - context = SpringApplication.run(TestDuplicateMapping.class, "--server.port=0"); + context = SpringApplication.run(TestMultipleMappingsWithReturnValue.class, "--server.port=0"); fail("Exception expected on duplicate mapping"); } - catch (BeanCreationException e) { - assertThat(e.getCause().getMessage()).startsWith("Duplicate @StreamListener mapping"); + catch (IllegalArgumentException e) { + assertThat(e.getMessage()).startsWith(StreamListenerErrorMessages.MULTIPLE_VALUE_RETURNING_METHODS); } finally { if (context != null) { @@ -72,16 +75,20 @@ public class StreamListenerDuplicateMappingTests { } } - @EnableBinding(Sink.class) + @EnableBinding(Processor.class) @EnableAutoConfiguration - public static class TestDuplicateMapping { + public static class TestMultipleMappingsWithReturnValue { - @StreamListener(Sink.INPUT) - public void receive(Message fooMessage) { + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public String receive(Message fooMessage) { + return null; } - @StreamListener(Sink.INPUT) - public void receiveDuplicateMapping(Message fooMessage) { + @StreamListener(Processor.INPUT) + @SendTo(Processor.OUTPUT) + public String receiveDuplicateMapping(Message fooMessage) { + return null; } } 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 new file mode 100644 index 000000000..50706f8c1 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerWithConditionsTest.java @@ -0,0 +1,141 @@ +/* + * Copyright 2016-2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.config; + +import java.util.ArrayList; +import java.util.List; +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; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binding.StreamListenerErrorMessages; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.handler.annotation.Payload; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; + +/** + * @author Marius Bogoevici + */ +public class StreamListenerWithConditionsTest { + + @Test + @SuppressWarnings("unchecked") + public void testAnnotatedArgumentsWithConditionalClass() throws Exception { + ConfigurableApplicationContext context = SpringApplication.run(TestPojoWithAnnotatedArguments.class, + "--server.port=0"); + + TestPojoWithAnnotatedArguments testPojoWithAnnotatedArguments = context + .getBean(TestPojoWithAnnotatedArguments.class); + Sink sink = context.getBean(Sink.class); + String id = UUID.randomUUID().toString(); + sink.input().send(MessageBuilder.withPayload("{\"foo\":\"barbar" + id + "\"}") + .setHeader("contentType", "application/json").setHeader("testHeader", "testValue") + .setHeader("type", "foo").build()); + sink.input().send(MessageBuilder.withPayload("{\"bar\":\"foofoo" + id + "\"}") + .setHeader("contentType", "application/json").setHeader("testHeader", "testValue") + .setHeader("type", "bar").build()); + sink.input().send(MessageBuilder.withPayload("{\"bar\":\"foofoo" + id + "\"}") + .setHeader("contentType", "application/json").setHeader("testHeader", "testValue") + .setHeader("type", "qux").build()); + assertThat(testPojoWithAnnotatedArguments.receivedFoo).hasSize(1); + assertThat(testPojoWithAnnotatedArguments.receivedFoo.get(0)).hasFieldOrPropertyWithValue("foo", + "barbar" + id); + assertThat(testPojoWithAnnotatedArguments.receivedBar).hasSize(1); + assertThat(testPojoWithAnnotatedArguments.receivedBar.get(0)).hasFieldOrPropertyWithValue("bar", + "foofoo" + id); + context.close(); + } + + @Test + @SuppressWarnings("unchecked") + public void testConditionalFailsWithReturnValue() throws Exception { + try { + ConfigurableApplicationContext context = SpringApplication.run(TestConditionalOnMethodWithReturnValueFails.class, + "--server.port=0"); + 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); + } + } + + @Test + @SuppressWarnings("unchecked") + public void testConditionalFailsWithDeclarativeMethod() throws Exception { + try { + ConfigurableApplicationContext context = SpringApplication.run(TestConditionalOnDeclarativeMethodFails.class, + "--server.port=0"); + context.close(); + fail("Context creation failure expected"); + } catch (BeanCreationException e) { + assertThat(e).hasRootCauseInstanceOf(IllegalArgumentException.class); + assertThat(e.getCause()).hasMessageContaining(StreamListenerErrorMessages.CONDITION_ON_DECLARATIVE_METHOD); + } + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + public static class TestPojoWithAnnotatedArguments { + + List receivedFoo = new ArrayList<>(); + + List receivedBar = new ArrayList<>(); + + @StreamListener(value = Sink.INPUT, condition = "headers['type']=='foo'") + public void receive(@Payload StreamListenerTestUtils.FooPojo fooPojo) { + this.receivedFoo.add(fooPojo); + } + + @StreamListener(target = Sink.INPUT, condition = "headers['type']=='bar'") + public void receive(@Payload StreamListenerTestUtils.BarPojo barPojo) { + this.receivedBar.add(barPojo); + } + + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + public static class TestConditionalOnDeclarativeMethodFails { + + @StreamListener(condition = "headers['type']=='foo'") + public void receive(@Input("input") MessageChannel input) { + // do nothing + } + } + + @EnableBinding(Sink.class) + @EnableAutoConfiguration + public static class TestConditionalOnMethodWithReturnValueFails { + + @StreamListener(value = Sink.INPUT, condition = "headers['type']=='foo'") + public String receive(String value) { + return null; + } + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamListener.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamListener.java index 0b4c9fc13..15834c082 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamListener.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/StreamListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,6 +23,7 @@ import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter; +import org.springframework.core.annotation.AliasFor; import org.springframework.messaging.handler.annotation.MessageMapping; /** @@ -130,7 +131,21 @@ public @interface StreamListener { /** * The name of the binding target (e.g. channel) that the method subscribes to. + * @return the name of the binding target. */ + @AliasFor("target") String value() default ""; + /** + * The name of the binding target (e.g. channel) that the method subscribes to. + * @return the name of the binding target. + */ + @AliasFor("value") + String target() default ""; + + /** + * A condition that must be met by all items that are dispatched to this method. + * @return a SpEL expression that must evaluate to a {@code boolean} value. + */ + String condition() default ""; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DispatchingStreamListenerMessageHandler.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DispatchingStreamListenerMessageHandler.java new file mode 100644 index 000000000..131ceb2e7 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/DispatchingStreamListenerMessageHandler.java @@ -0,0 +1,125 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binding; + +import java.util.ArrayList; +import java.util.Collection; + +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; +import org.springframework.util.Assert; + +/** + * An {@link AbstractReplyProducingMessageHandler} that delegates to a + * collection of internal {@link ConditionalStreamListenerHandler} instances, + * executing the ones that match the given expression. + * + * @author Marius Bogoevici + * @since 1.2 + */ +final class DispatchingStreamListenerMessageHandler extends AbstractReplyProducingMessageHandler { + + private final Collection handlerMethods; + + private final EvaluationContext evaluationContext; + + DispatchingStreamListenerMessageHandler(Collection handlerMethods, + EvaluationContext evaluationContext) { + Assert.notEmpty(handlerMethods, "'handlerMethods' cannot be empty"); + Assert.notNull(evaluationContext, "'evaluationContext' cannot be empty"); + this.handlerMethods = handlerMethods; + this.evaluationContext = evaluationContext; + } + + @Override + protected boolean shouldCopyRequestHeaders() { + return false; + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + Collection matchingHandlers = findMatchingHandlers(requestMessage); + if (matchingHandlers.size() == 0) { + if (logger.isWarnEnabled()) { + logger.warn("Cannot find a @StreamListener matching for message with id: " + + requestMessage.getHeaders().getId()); + } + return null; + } + else if (matchingHandlers.size() > 1) { + for (ConditionalStreamListenerHandler matchingMethod : matchingHandlers) { + matchingMethod.handleMessage(requestMessage); + } + return null; + } + else { + final ConditionalStreamListenerHandler singleMatchingHandler = matchingHandlers.iterator().next(); + singleMatchingHandler.handleMessage(requestMessage); + return null; + } + } + + private Collection findMatchingHandlers(Message message) { + ArrayList matchingMethods = new ArrayList<>(); + for (ConditionalStreamListenerHandler conditionalStreamListenerHandlerMethod : this.handlerMethods) { + if (conditionalStreamListenerHandlerMethod.getCondition() == null) { + matchingMethods.add(conditionalStreamListenerHandlerMethod); + } + else { + boolean conditionMetOnMessage = conditionalStreamListenerHandlerMethod.getCondition().getValue( + this.evaluationContext, message, Boolean.class); + if (conditionMetOnMessage) { + matchingMethods.add(conditionalStreamListenerHandlerMethod); + } + } + } + return matchingMethods; + } + + static class ConditionalStreamListenerHandler implements MessageHandler { + + private Expression condition; + + private StreamListenerMessageHandler streamListenerMessageHandler; + + ConditionalStreamListenerHandler(Expression condition, + StreamListenerMessageHandler streamListenerMessageHandler) { + Assert.notNull(streamListenerMessageHandler, "the message handler cannot be null"); + Assert.isTrue(condition == null || streamListenerMessageHandler.isVoid(), + "cannot specify a condition and a return value at the same time"); + this.condition = condition; + this.streamListenerMessageHandler = streamListenerMessageHandler; + } + + public Expression getCondition() { + return condition; + } + + public boolean isVoid() { + return this.streamListenerMessageHandler.isVoid(); + } + + @Override + public void handleMessage(Message message) throws MessagingException { + this.streamListenerMessageHandler.handleMessage(message); + } + } +} 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 cc80f3fa4..5b09089ce 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,34 +18,46 @@ package org.springframework.cloud.stream.binding; import java.lang.reflect.Method; import java.util.ArrayList; -import java.util.HashMap; +import java.util.Collection; import java.util.List; import java.util.Map; import org.springframework.aop.framework.Advised; import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanFactoryUtils; import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.beans.factory.InitializingBean; 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; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.Output; import org.springframework.cloud.stream.annotation.StreamListener; 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.AnnotationUtils; -import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; -import org.springframework.messaging.Message; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessagingException; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.core.DestinationResolver; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.util.Assert; +import org.springframework.util.LinkedMultiValueMap; +import org.springframework.util.MultiValueMap; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; @@ -57,13 +69,20 @@ import org.springframework.util.StringUtils; * @author Ilayaperumal Gopinathan */ public class StreamListenerAnnotationBeanPostProcessor - implements BeanPostProcessor, ApplicationContextAware, SmartInitializingSingleton { + implements BeanPostProcessor, ApplicationContextAware, BeanFactoryAware, SmartInitializingSingleton, + InitializingBean { - private final DestinationResolver binderAwareChannelResolver; + private static final SpelExpressionParser SPEL_EXPRESSION_PARSER = new SpelExpressionParser(); - private final MessageHandlerMethodFactory messageHandlerMethodFactory; + @Autowired + @Lazy + private DestinationResolver binderAwareChannelResolver; - private final Map mappedBindings = new HashMap<>(); + @Autowired + @Lazy + private MessageHandlerMethodFactory messageHandlerMethodFactory; + + private final MultiValueMap mappedListenerMethods = new LinkedMultiValueMap<>(); private ConfigurableApplicationContext applicationContext; @@ -71,25 +90,38 @@ public class StreamListenerAnnotationBeanPostProcessor private final List> streamListenerResultAdapters = new ArrayList<>(); - public StreamListenerAnnotationBeanPostProcessor(DestinationResolver binderAwareChannelResolver, - MessageHandlerMethodFactory messageHandlerMethodFactory) { - Assert.notNull(binderAwareChannelResolver, "Destination resolver cannot be null"); - Assert.notNull(messageHandlerMethodFactory, "Message handler method factory cannot be null"); - this.binderAwareChannelResolver = binderAwareChannelResolver; - this.messageHandlerMethodFactory = messageHandlerMethodFactory; - } + private EvaluationContext evaluationContext; + + private BeanFactory beanFactory; + + private BeanExpressionResolver resolver; + + private BeanExpressionContext expressionContext; @Override @SuppressWarnings({ "rawtypes", "unchecked" }) - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + public final void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = (ConfigurableApplicationContext) applicationContext; - Map parameterAdapterMap = - BeanFactoryUtils.beansOfTypeIncludingAncestors(this.applicationContext, StreamListenerParameterAdapter.class); + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + if (beanFactory instanceof ConfigurableListableBeanFactory) { + this.resolver = ((ConfigurableListableBeanFactory) beanFactory).getBeanExpressionResolver(); + this.expressionContext = new BeanExpressionContext((ConfigurableListableBeanFactory) beanFactory, null); + } + } + + @Override + public void afterPropertiesSet() throws Exception { + Map parameterAdapterMap = BeanFactoryUtils + .beansOfTypeIncludingAncestors(this.applicationContext, StreamListenerParameterAdapter.class); for (StreamListenerParameterAdapter parameterAdapter : parameterAdapterMap.values()) { this.streamListenerParameterAdapters.add(parameterAdapter); } - Map resultAdapterMap = - BeanFactoryUtils.beansOfTypeIncludingAncestors(this.applicationContext, StreamListenerResultAdapter.class); + Map resultAdapterMap = BeanFactoryUtils + .beansOfTypeIncludingAncestors(this.applicationContext, StreamListenerResultAdapter.class); this.streamListenerResultAdapters.add(new MessageChannelStreamListenerResultAdapter()); for (StreamListenerResultAdapter resultAdapter : resultAdapterMap.values()) { this.streamListenerResultAdapters.add(resultAdapter); @@ -97,18 +129,19 @@ public class StreamListenerAnnotationBeanPostProcessor } @Override - public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + public final Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { return bean; } @Override - public Object postProcessAfterInitialization(final Object bean, String beanName) throws BeansException { + public final Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException { Class targetClass = AopUtils.isAopProxy(bean) ? AopUtils.getTargetClass(bean) : bean.getClass(); ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() { @Override public void doWith(final Method method) throws IllegalArgumentException, IllegalAccessException { StreamListener streamListener = AnnotationUtils.findAnnotation(method, StreamListener.class); if (streamListener != null && !method.isBridge()) { + streamListener = postProcessAnnotation(streamListener, method); Assert.isTrue(method.getAnnotation(Input.class) == null, StreamListenerErrorMessages.INPUT_AT_STREAM_LISTENER); String methodAnnotatedInboundName = streamListener.value(); @@ -119,7 +152,7 @@ public class StreamListenerAnnotationBeanPostProcessor methodAnnotatedOutboundName); StreamListenerMethodUtils.validateStreamListenerMethod(method, inputAnnotationCount, outputAnnotationCount, methodAnnotatedInboundName, methodAnnotatedOutboundName, - isDeclarative); + isDeclarative, streamListener.condition()); if (!method.getReturnType().equals(Void.TYPE)) { if (!StringUtils.hasText(methodAnnotatedOutboundName)) { if (outputAnnotationCount == 0) { @@ -143,6 +176,18 @@ public class StreamListenerAnnotationBeanPostProcessor return bean; } + /** + * Extension point, allowing subclasses to customize the {@link StreamListener} annotation detected by + * the postprocessor. + * + * @param originalAnnotation the original annotation + * @param annotatedMethod the method on which the annotation has been found + * @return the postprocessed {@link StreamListener} annotation + */ + protected StreamListener postProcessAnnotation(StreamListener originalAnnotation, Method annotatedMethod) { + return originalAnnotation; + } + private boolean checkDeclarativeMethod(Method method, String methodAnnotatedInboundName, String methodAnnotatedOutboundName) { int methodArgumentsLength = method.getParameterTypes().length; @@ -232,7 +277,7 @@ public class StreamListenerAnnotationBeanPostProcessor } } try { - if (method.getReturnType().equals(Void.TYPE)) { + if (Void.TYPE.equals(method.getReturnType())) { method.invoke(bean, arguments); } else { @@ -260,24 +305,13 @@ public class StreamListenerAnnotationBeanPostProcessor } } - protected void registerHandlerMethodOnListenedChannel(Method method, StreamListener streamListener, Object bean) { - Method targetMethod = checkProxy(method, bean); + protected final void registerHandlerMethodOnListenedChannel(Method method, StreamListener streamListener, Object bean) { Assert.hasText(streamListener.value(), "The binding name cannot be null"); - final InvocableHandlerMethod invocableHandlerMethod = this.messageHandlerMethodFactory - .createInvocableHandlerMethod(bean, targetMethod); if (!StringUtils.hasText(streamListener.value())) { throw new BeanInitializationException("A bound component name must be specified"); } - if (this.mappedBindings.containsKey(streamListener.value())) { - throw new BeanInitializationException("Duplicate @" + StreamListener.class.getSimpleName() - + " mapping for '" + streamListener.value() + "' on " + invocableHandlerMethod.getShortLogMessage() - + " already existing for " + this.mappedBindings.get(streamListener.value()).getShortLogMessage()); - } - this.mappedBindings.put(streamListener.value(), invocableHandlerMethod); - SubscribableChannel channel = this.applicationContext.getBean(streamListener.value(), - SubscribableChannel.class); final String defaultOutputChannel = StreamListenerMethodUtils.getOutboundBindingTargetName(method); - if (invocableHandlerMethod.isVoid()) { + 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"); } @@ -286,22 +320,52 @@ public class StreamListenerAnnotationBeanPostProcessor "An output channel must be specified for a method that can return a value"); } StreamListenerMethodUtils.validateStreamListenerMessageHandler(method); - StreamListenerMessageHandler handler = new StreamListenerMessageHandler(invocableHandlerMethod); - handler.setApplicationContext(this.applicationContext); - handler.setChannelResolver(this.binderAwareChannelResolver); - if (!StringUtils.isEmpty(defaultOutputChannel)) { - handler.setOutputChannelName(defaultOutputChannel); - } - handler.afterPropertiesSet(); - channel.subscribe(handler); + mappedListenerMethods.add(streamListener.value(), + new StreamListenerHandlerMethodMapping(bean, method, streamListener.condition(), defaultOutputChannel)); } @Override - public void afterSingletonsInstantiated() { - // Dump the mappings after the context has been created, ensuring that beans can - // be processed correctly - // again. - this.mappedBindings.clear(); + public final void afterSingletonsInstantiated() { + this.evaluationContext = IntegrationContextUtils.getEvaluationContext(this.applicationContext.getBeanFactory()); + for (Map.Entry> mappedBindingEntry : mappedListenerMethods + .entrySet()) { + Collection 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); + 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()); + Expression condition = SPEL_EXPRESSION_PARSER.parseExpression(conditionAsString); + handlers.add(new DispatchingStreamListenerMessageHandler.ConditionalStreamListenerHandler( + condition, streamListenerMessageHandler)); + } + else { + handlers.add(new DispatchingStreamListenerMessageHandler.ConditionalStreamListenerHandler( + null, streamListenerMessageHandler)); + } + } + if (handlers.size() > 1) { + for (DispatchingStreamListenerMessageHandler.ConditionalStreamListenerHandler handler : handlers) { + Assert.isTrue(handler.isVoid(), StreamListenerErrorMessages.MULTIPLE_VALUE_RETURNING_METHODS); + } + } + DispatchingStreamListenerMessageHandler handler = new DispatchingStreamListenerMessageHandler( + handlers, this.evaluationContext); + 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) { @@ -337,33 +401,70 @@ public class StreamListenerAnnotationBeanPostProcessor return method; } - private final class StreamListenerMessageHandler extends AbstractReplyProducingMessageHandler { + private String resolveExpressionAsString(String value) { + Object resolved = resolveExpression(value); + if (resolved instanceof String) { + return (String) resolved; + } + else { + throw new IllegalStateException("Resolved to [" + resolved.getClass() + "] for [" + value + "]"); + } + } - private final InvocableHandlerMethod invocableHandlerMethod; + private Object resolveExpression(String value) { + String resolvedValue = resolve(value); - private StreamListenerMessageHandler(InvocableHandlerMethod invocableHandlerMethod) { - this.invocableHandlerMethod = invocableHandlerMethod; + if (!(resolvedValue.startsWith("#{") && value.endsWith("}"))) { + return resolvedValue; } - @Override - protected boolean shouldCopyRequestHeaders() { - return false; + return this.resolver.evaluate(resolvedValue, this.expressionContext); + } + + /** + * Resolve the specified value if possible. + * + * @see ConfigurableBeanFactory#resolveEmbeddedValue + */ + private String resolve(String value) { + if (this.beanFactory != null && this.beanFactory instanceof ConfigurableBeanFactory) { + return ((ConfigurableBeanFactory) this.beanFactory).resolveEmbeddedValue(value); + } + return value; + } + + private class StreamListenerHandlerMethodMapping { + + private Object targetBean; + + private Method method; + + private String condition; + + private String defaultOutputChannel; + + StreamListenerHandlerMethodMapping(Object targetBean, Method method, String condition, + String defaultOutputChannel) { + this.targetBean = targetBean; + this.method = method; + this.condition = condition; + this.defaultOutputChannel = defaultOutputChannel; } - @Override - protected Object handleRequestMessage(Message requestMessage) { - try { - return this.invocableHandlerMethod.invoke(requestMessage); - } - catch (Exception e) { - if (e instanceof MessagingException) { - throw (MessagingException) e; - } - else { - throw new MessagingException(requestMessage, - "Exception thrown while invoking " + this.invocableHandlerMethod.getShortLogMessage(), e); - } - } + Object getTargetBean() { + return targetBean; + } + + Method getMethod() { + return method; + } + + String getCondition() { + return condition; + } + + String getDefaultOutputChannel() { + return defaultOutputChannel; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerErrorMessages.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerErrorMessages.java index 5bc6d1b1b..5f8c927a7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerErrorMessages.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerErrorMessages.java @@ -60,4 +60,10 @@ public abstract class StreamListenerErrorMessages { public static final String INVALID_OUTPUT_VALUES = "Cannot set both output (@Output/@SendTo) method annotation value" + " and @Output annotation as a method parameter"; + + public static final String CONDITION_ON_DECLARATIVE_METHOD = "Cannot set a condition when using @StreamListener in declarative mode"; + + public static final String CONDITION_ON_METHOD_RETURNING_VALUE = "Cannot set a condition for methods that return a value"; + + public static final String MULTIPLE_VALUE_RETURNING_METHODS = "If multiple @StreamListener methods are listening to the same binding target, none of them may return a value"; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerMessageHandler.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerMessageHandler.java new file mode 100644 index 000000000..6a7d4ee8f --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerMessageHandler.java @@ -0,0 +1,60 @@ +/* + * Copyright 2017 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binding; + +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; + +/** + * @author Marius Bogoevici + * @since 1.2 + */ +public class StreamListenerMessageHandler extends AbstractReplyProducingMessageHandler { + + private final InvocableHandlerMethod invocableHandlerMethod; + + StreamListenerMessageHandler(InvocableHandlerMethod invocableHandlerMethod) { + this.invocableHandlerMethod = invocableHandlerMethod; + } + + @Override + protected boolean shouldCopyRequestHeaders() { + return false; + } + + public boolean isVoid() { + return invocableHandlerMethod.isVoid(); + } + + @Override + protected Object handleRequestMessage(Message requestMessage) { + try { + return this.invocableHandlerMethod.invoke(requestMessage); + } + catch (Exception e) { + if (e instanceof MessagingException) { + throw (MessagingException) e; + } + else { + throw new MessagingException(requestMessage, + "Exception thrown while invoking " + this.invocableHandlerMethod.getShortLogMessage(), e); + } + } + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerMethodUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerMethodUtils.java index 364d88abe..ef3ace67d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerMethodUtils.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerMethodUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -61,7 +61,7 @@ public class StreamListenerMethodUtils { protected static void validateStreamListenerMethod(Method method, int inputAnnotationCount, int outputAnnotationCount, String methodAnnotatedInboundName, String methodAnnotatedOutboundName, - boolean isDeclarative) { + boolean isDeclarative, String condition) { int methodArgumentsLength = method.getParameterTypes().length; if (!isDeclarative) { Assert.isTrue(inputAnnotationCount == 0 && outputAnnotationCount == 0, @@ -82,7 +82,13 @@ public class StreamListenerMethodUtils { if (StringUtils.hasText(methodAnnotatedOutboundName)) { Assert.isTrue(outputAnnotationCount == 0, StreamListenerErrorMessages.INVALID_OUTPUT_VALUES); } + if (!Void.TYPE.equals(method.getReturnType())) { + Assert.isTrue(!StringUtils.hasText(condition), + StreamListenerErrorMessages.CONDITION_ON_METHOD_RETURNING_VALUE); + } if (isDeclarative) { + Assert.isTrue(!StringUtils.hasText(condition), + StreamListenerErrorMessages.CONDITION_ON_DECLARATIVE_METHOD); for (int parameterIndex = 0; parameterIndex < methodArgumentsLength; parameterIndex++) { MethodParameter methodParameter = MethodParameter.forMethodOrConstructor(method, parameterIndex); if (methodParameter.hasParameterAnnotation(Input.class)) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 7ca343337..58cf37119 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -52,7 +52,6 @@ import org.springframework.cloud.stream.converter.CompositeMessageConverterFacto import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.DependsOn; -import org.springframework.context.annotation.Lazy; import org.springframework.expression.PropertyAccessor; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.config.IntegrationEvaluationContextFactoryBean; @@ -81,6 +80,8 @@ public class BindingServiceConfiguration { private static final String ERROR_CHANNEL_NAME = "error"; + public static final String STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME = "streamListenerAnnotationBeanPostProcessor"; + @Autowired(required = false) private ObjectMapper objectMapper; @@ -186,14 +187,6 @@ public class BindingServiceConfiguration { return messageHandlerMethodFactory; } - @Bean - @ConditionalOnMissingBean(StreamListenerAnnotationBeanPostProcessor.class) - public static StreamListenerAnnotationBeanPostProcessor bindToAnnotationBeanPostProcessor( - @Lazy BinderAwareChannelResolver binderAwareChannelResolver, - @Lazy MessageHandlerMethodFactory messageHandlerMethodFactory) { - return new StreamListenerAnnotationBeanPostProcessor(binderAwareChannelResolver, messageHandlerMethodFactory); - } - @Bean // provided for backwards compatibility scenarios public ChannelBindingServiceProperties channelBindingServiceProperties( @@ -201,6 +194,11 @@ public class BindingServiceConfiguration { return new ChannelBindingServiceProperties(bindingServiceProperties); } + @Bean(name = STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME) + public static StreamListenerAnnotationBeanPostProcessor streamListenerAnnotationBeanPostProcessor() { + return new StreamListenerAnnotationBeanPostProcessor(); + } + // IMPORTANT: Nested class to avoid instantiating all of the above early @Configuration protected static class PostProcessorConfiguration {