GH-1535 Added support for Binding auto-startup

Resolves #1535

docs/src/main/asciidoc/spring-cloud-stream.adoc

@@ -1404,6 +1404,10 @@ The following binding properties are available for input bindings only and must
 
 Default values can be set by using the `spring.cloud.stream.default.consumer` prefix (for example, `spring.cloud.stream.default.consumer.headerMode=none`).
 
+autoStartup::
+Signals if this consumer needs to be started automatically
++
+Default: `true`.
 concurrency::
 The concurrency of the inbound consumer.
 +
@@ -1478,6 +1482,10 @@ The following binding properties are available for output bindings only and must
 
 Default values can be set by using the prefix `spring.cloud.stream.default.producer` (for example, `spring.cloud.stream.default.producer.partitionKeyExpression=payload.id`).
 
+autoStartup::
+Signals if this consumer needs to be started automatically
++
+Default: `true`.
 partitionKeyExpression::
 A SpEL expression that determines how to partition outbound data.
 If set, or if `partitionKeyExtractorClass` is set, outbound data on this channel is partitioned. `partitionCount` must be set to a value greater than 1 to be effective.

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -190,7 +190,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				throw new BinderException("Exception thrown while building outbound endpoint", e);
 			}
 		}
-		if (producerMessageHandler instanceof Lifecycle) {
+		if (producerProperties.isAutoStartup() && producerMessageHandler instanceof Lifecycle) {
 			((Lifecycle) producerMessageHandler).start();
 		}
 		this.postProcessOutputChannel(outputChannel, producerProperties);
@@ -361,7 +361,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			if (consumerEndpoint instanceof InitializingBean) {
 				((InitializingBean) consumerEndpoint).afterPropertiesSet();
 			}
-			if (consumerEndpoint instanceof Lifecycle) {
+			if (properties.isAutoStartup() && consumerEndpoint instanceof Lifecycle) {
 				((Lifecycle) consumerEndpoint).start();
 			}
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java

@@ -36,6 +36,13 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public class ConsumerProperties {
 
+	/**
+	 * Signals if this consumer needs to be started automatically
+	 *
+	 * Default: true
+	 */
+	private boolean autoStartup = true;
+
 	/**
 	 * The concurrency setting of the consumer. Default: 1.
 	 */
@@ -261,4 +268,12 @@ public class ConsumerProperties {
 	public void setMultiplex(boolean multiplex) {
 		this.multiplex = multiplex;
 	}
+
+	public boolean isAutoStartup() {
+		return autoStartup;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.java

@@ -41,6 +41,13 @@ import org.springframework.expression.Expression;
 @JsonInclude(Include.NON_DEFAULT)
 public class ProducerProperties {
 
+	/**
+	 * Signals if this producer needs to be started automatically
+	 *
+	 * Default: true
+	 */
+	private boolean autoStartup = true;
+
 	@JsonSerialize(using = ExpressionSerializer.class)
 	private Expression partitionKeyExpression;
 
@@ -193,6 +200,14 @@ public class ProducerProperties {
 		this.partitionSelectorName = partitionSelectorName;
 	}
 
+	public boolean isAutoStartup() {
+		return autoStartup;
+	}
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
 	static class ExpressionSerializer extends JsonSerializer<Expression> {
 
 		@Override

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java

@@ -16,10 +16,12 @@
 
 package org.springframework.cloud.stream.binding;
 
+import java.lang.reflect.Field;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
@@ -37,7 +39,9 @@ import org.mockito.stubbing.Answer;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.annotation.Input;
@@ -54,22 +58,31 @@ import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
 import org.springframework.cloud.stream.config.BindingProperties;
 import org.springframework.cloud.stream.config.BindingServiceConfiguration;
 import org.springframework.cloud.stream.config.BindingServiceProperties;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.cloud.stream.messaging.Sink;
 import org.springframework.cloud.stream.reflection.GenericsUtils;
 import org.springframework.cloud.stream.utils.MockBinderConfiguration;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.test.util.TestUtils;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.SubscribableChannel;
 import org.springframework.messaging.core.DestinationResolutionException;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.ReflectionUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -611,6 +624,30 @@ public class BindingServiceTests {
 		scheduler.destroy();
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	public void testBindingAutostartup() throws Exception {
+		ApplicationContext context = new SpringApplicationBuilder(FooConfiguration.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.input.consumer.auto-startup=false");
+		BindingService bindingService = context.getBean(BindingService.class);
+
+		Field cbField = ReflectionUtils.findField(BindingService.class, "consumerBindings");
+		cbField.setAccessible(true);
+		Map<String, Object> cbMap = (Map<String, Object>) cbField.get(bindingService);
+		Binding<?> inputBinding = ((List<Binding<?>>)cbMap.get("input")).get(0);
+		assertFalse(inputBinding.isRunning());
+	}
+
+	@EnableBinding(Sink.class)
+	@Import(TestChannelBinderConfiguration.class)
+	@EnableAutoConfiguration
+	public static class FooConfiguration {
+		@ServiceActivator(inputChannel=Processor.INPUT)
+		public void echo(Message<?> value) throws Exception {
+		}
+	}
+
 	private DefaultBinderFactory createMockBinderFactory() {
 		BinderTypeRegistry binderTypeRegistry = createMockBinderTypeRegistry();
 		return new DefaultBinderFactory(