diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index bd64838f4..1f4d223a1 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/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. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 9849c0f86..ecb33c249 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -190,7 +190,7 @@ public abstract class AbstractMessageChannelBinder { @Override diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index 609a41742..99a850990 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/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 cbMap = (Map) cbField.get(bindingService); + Binding inputBinding = ((List>)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(