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 {