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;

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<StreamListenerTestUtils.FooPojo> 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<String,Object> attributes = new HashMap<>(AnnotationUtils.getAnnotationAttributes(originalAnnotation));
+					attributes.put("condition", "headers['type']=='" + originalAnnotation.condition() + "'");
+					return AnnotationUtils.synthesizeAnnotation(attributes, StreamListener.class, annotatedMethod);
+				}
+			};
+		}
+	}
+}

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<String> fooMessage) {
+		@StreamListener(Processor.INPUT)
+		@SendTo(Processor.OUTPUT)
+		public String receive(Message<String> fooMessage) {
+			return null;
 		}
 
-		@StreamListener(Sink.INPUT)
-		public void receiveDuplicateMapping(Message<String> fooMessage) {
+		@StreamListener(Processor.INPUT)
+		@SendTo(Processor.OUTPUT)
+		public String receiveDuplicateMapping(Message<String> fooMessage) {
+			return null;
 		}
 	}
 

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<StreamListenerTestUtils.FooPojo> receivedFoo = new ArrayList<>();
+
+		List<StreamListenerTestUtils.BarPojo> 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;
+		}
+	}
+}

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 "";
 }

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<ConditionalStreamListenerHandler> handlerMethods;
+
+	private final EvaluationContext evaluationContext;
+
+	DispatchingStreamListenerMessageHandler(Collection<ConditionalStreamListenerHandler> 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<ConditionalStreamListenerHandler> 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<ConditionalStreamListenerHandler> findMatchingHandlers(Message<?> message) {
+		ArrayList<ConditionalStreamListenerHandler> 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);
+		}
+	}
+}

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<MessageChannel> binderAwareChannelResolver;
+	private static final SpelExpressionParser SPEL_EXPRESSION_PARSER = new SpelExpressionParser();
 
-	private final MessageHandlerMethodFactory messageHandlerMethodFactory;
+	@Autowired
+	@Lazy
+	private DestinationResolver<MessageChannel> binderAwareChannelResolver;
 
-	private final Map<String, InvocableHandlerMethod> mappedBindings = new HashMap<>();
+	@Autowired
+	@Lazy
+	private MessageHandlerMethodFactory messageHandlerMethodFactory;
+
+	private final MultiValueMap<String, StreamListenerHandlerMethodMapping> mappedListenerMethods = new LinkedMultiValueMap<>();
 
 	private ConfigurableApplicationContext applicationContext;
 
@@ -71,25 +90,38 @@ public class StreamListenerAnnotationBeanPostProcessor
 
 	private final List<StreamListenerResultAdapter<?, ?>> streamListenerResultAdapters = new ArrayList<>();
 
-	public StreamListenerAnnotationBeanPostProcessor(DestinationResolver<MessageChannel> 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<String, StreamListenerParameterAdapter> 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<String, StreamListenerParameterAdapter> parameterAdapterMap = BeanFactoryUtils
+				.beansOfTypeIncludingAncestors(this.applicationContext, StreamListenerParameterAdapter.class);
 		for (StreamListenerParameterAdapter parameterAdapter : parameterAdapterMap.values()) {
 			this.streamListenerParameterAdapters.add(parameterAdapter);
 		}
-		Map<String, StreamListenerResultAdapter> resultAdapterMap =
-				BeanFactoryUtils.beansOfTypeIncludingAncestors(this.applicationContext, StreamListenerResultAdapter.class);
+		Map<String, StreamListenerResultAdapter> 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<String, List<StreamListenerHandlerMethodMapping>> mappedBindingEntry : mappedListenerMethods
+				.entrySet()) {
+			Collection<DispatchingStreamListenerMessageHandler.ConditionalStreamListenerHandler> 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;
 		}
 	}
 

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";
 }

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);
+			}
+		}
+	}
+}

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)) {

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 {