From b3a0f2c2e1ab98581f3303ed5d6cc3864adcc1e7 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Thu, 1 Dec 2016 18:33:22 -0500 Subject: [PATCH] Add support for defaults at binding level Fixes #446 Supports properties prefixed with `spring.cloud.stream.default` for specifying default values for large number of channels or for dynamically generated channels. Note: this does not cover binding specific properties, as the property objects are created under binder's control but the same technique can be applied there. Rework default property binding Minor: javadoc Updated documentation --- .../spring-cloud-stream-overview.adoc | 16 +++- .../ChannelBindingServiceProperties.java | 35 +-------- .../EnvironmentEntryInitializingTreeMap.java | 75 +++++++++++++++++++ ...SourceBindingWithGlobalPropertiesTest.java | 65 ++++++++++++++++ 4 files changed, 156 insertions(+), 35 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnvironmentEntryInitializingTreeMap.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 540f9ec8c..4950ccfc4 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -885,17 +885,21 @@ Default: false. Binding properties are supplied using the format `spring.cloud.stream.bindings..=`. The `` represents the name of the channel being configured (e.g., `output` for a `Source`). +To avoid repetition, Spring Cloud Stream supports setting values for all channels, in the format `spring.cloud.stream.default.=`. + In what follows, we indicate where we have omitted the `spring.cloud.stream.bindings..` prefix and focus just on the property name, with the understanding that the prefix will be included at runtime. ==== Properties for Use of Spring Cloud Stream -The following binding properties are available for both input and output bindings and -must be prefixed with `spring.cloud.stream.bindings..`. +The following binding properties are available for both input and output bindings and must be prefixed with `spring.cloud.stream.bindings..`, e.g. `spring.cloud.stream.bindings.input.destination=ticktock`. + +Default values can be set by using the prefix `spring.cloud.stream.default`, e.g. `spring.cloud.stream.default.contentType=application/json`. destination:: The target destination of a channel on the bound middleware (e.g., the RabbitMQ exchange or Kafka topic). If the channel is bound as a consumer, it could be bound to multiple destinations and the destination names can be specified as comma separated String values. If not set, the channel name is used instead. + The default value of this property cannot be overridden. group:: The consumer group of the channel. Applies only to inbound bindings. @@ -915,7 +919,9 @@ Default: null (the default binder will be used, if one exists). ==== Consumer properties -The following binding properties are available for input bindings only and must be prefixed with `spring.cloud.stream.bindings..consumer.`. +The following binding properties are available for input bindings only and must be prefixed with `spring.cloud.stream.bindings..consumer.`, e.g. `spring.cloud.stream.bindings.input.consumer.concurrency=3`. + +Default values can be set by using the prefix `spring.cloud.stream.default.consumer`, e.g. `spring.cloud.stream.default.consumer.headerMode=raw`. concurrency:: The concurrency of the inbound consumer. @@ -960,7 +966,9 @@ Default: `-1`. ==== Producer Properties -The following binding properties are available for output bindings only and must be prefixed with `spring.cloud.stream.bindings..producer.`. +The following binding properties are available for output bindings only and must be prefixed with `spring.cloud.stream.bindings..producer.`, e.g. `spring.cloud.stream.bindings.input.producer.partitionKeyExpression=payload.id`. + +Default values can be set by using the prefix `spring.cloud.stream.default.producer`, e.g. `spring.cloud.stream.default.producer.partitionKeyExpression=payload.id`. partitionKeyExpression:: A SpEL expression that determines how to partition outbound data. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java index 7067046da..6b5b7493d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java @@ -18,7 +18,6 @@ package org.springframework.cloud.stream.config; import java.util.HashMap; import java.util.Map; -import java.util.Properties; import java.util.TreeMap; import com.fasterxml.jackson.annotation.JsonInclude; @@ -59,16 +58,10 @@ public class ChannelBindingServiceProperties implements ApplicationContextAware, private Map binders = new HashMap<>(); - private Properties consumerDefaults = new Properties(); - - private Properties producerDefaults = new Properties(); - private String defaultBinder; private String[] dynamicDestinations = new String[0]; - private boolean ignoreUnknownProperties = true; - private ConfigurableApplicationContext applicationContext; public Map getBindings() { @@ -119,33 +112,12 @@ public class ChannelBindingServiceProperties implements ApplicationContextAware, this.dynamicDestinations = dynamicDestinations; } - public Properties getConsumerDefaults() { - return this.consumerDefaults; - } - - public void setConsumerDefaults(Properties consumerDefaults) { - this.consumerDefaults = consumerDefaults; - } - - public Properties getProducerDefaults() { - return this.producerDefaults; - } - - public void setProducerDefaults(Properties producerDefaults) { - this.producerDefaults = producerDefaults; - } - - public boolean isIgnoreUnknownProperties() { - return this.ignoreUnknownProperties; - } - - public void setIgnoreUnknownProperties(boolean ignoreUnknownProperties) { - this.ignoreUnknownProperties = ignoreUnknownProperties; - } - @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = (ConfigurableApplicationContext) applicationContext; + // override the bindings store with the environment-initializing version if in a Spring context + this.bindings = new EnvironmentEntryInitializingTreeMap<>(this.applicationContext, BindingProperties.class, + "spring.cloud.stream.default", new TreeMap(String.CASE_INSENSITIVE_ORDER)); } public void setConversionService(ConversionService conversionService) { @@ -226,4 +198,5 @@ public class ChannelBindingServiceProperties implements ApplicationContextAware, public String getBindingDestination(String channelName) { return getBindingProperties(channelName).getDestination(); } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnvironmentEntryInitializingTreeMap.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnvironmentEntryInitializingTreeMap.java new file mode 100644 index 000000000..37b7c6a87 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnvironmentEntryInitializingTreeMap.java @@ -0,0 +1,75 @@ +package org.springframework.cloud.stream.config; + +import java.util.AbstractMap; +import java.util.Map; +import java.util.Set; + +import org.springframework.beans.BeanUtils; +import org.springframework.boot.bind.PropertySourcesPropertyValues; +import org.springframework.boot.bind.RelaxedDataBinder; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.util.Assert; + +/** + * A {@link Map} implementation that initializes its entries by binding values from the + * supplied application context's environment. Any call to 'get()' will result in either + * returning the existing value or initializing a new entry by binding properties with the + * specified prefix from the environment. + * + * This is strictly intended to be used for configuration property values and not to be + * used as a general purpose map. + * + * This implementation is not thread safe. + * + * @author Marius Bogoevici + */ +public class EnvironmentEntryInitializingTreeMap extends AbstractMap { + + private final ConfigurableApplicationContext applicationContext; + + private final Class entryClass; + + private final String defaultsPrefix; + + private final Map delegate; + + /** + * Constructs the map. + * + * @param applicationContext the application context that supplies the property values + * @param entryClass the entry class + * @param defaultsPrefix the prefix for initializing the properties + * @param delegate the actual map that stores the values + */ + public EnvironmentEntryInitializingTreeMap(ConfigurableApplicationContext applicationContext, Class entryClass, + String defaultsPrefix, Map delegate) { + Assert.notNull(applicationContext, "The context cannot be null"); + Assert.notNull(entryClass, "The entry class cannot be null"); + Assert.notNull(defaultsPrefix, "The prefix for the property defaults cannot be null"); + Assert.notNull(delegate, "The delegate cannot be null"); + this.applicationContext = applicationContext; + this.entryClass = entryClass; + this.defaultsPrefix = defaultsPrefix; + this.delegate = delegate; + } + + @Override + public T get(Object key) { + if (!this.delegate.containsKey(key) && key instanceof String) { + T entry = BeanUtils.instantiate(entryClass); + if (applicationContext != null) { + RelaxedDataBinder defaultsDataBinder = new RelaxedDataBinder(entry, defaultsPrefix); + defaultsDataBinder.bind( + new PropertySourcesPropertyValues(applicationContext.getEnvironment().getPropertySources())); + } + this.delegate.put((String) key, entry); + } + return this.delegate.get(key); + } + + @Override + public Set> entrySet() { + return delegate.entrySet(); + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java new file mode 100644 index 000000000..50345b4fe --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java @@ -0,0 +1,65 @@ +/* + * Copyright 2015 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.binder; + +import org.assertj.core.api.Assertions; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.cloud.stream.config.ChannelBindingServiceProperties; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; +import org.springframework.context.annotation.Import; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringBootTest(classes = SourceBindingWithGlobalPropertiesTest.TestSource.class, properties = { + "spring.cloud.stream.default.contentType=application/json", + "spring.cloud.stream.bindings.output.destination=ticktock", + "spring.cloud.stream.default.producer.requiredGroups=someGroup", + "spring.cloud.stream.bindings.output.producer.headerMode=raw"}) +public class SourceBindingWithGlobalPropertiesTest { + + @Autowired + private ChannelBindingServiceProperties channelBindingServiceProperties; + + @SuppressWarnings("unchecked") + @Test + public void testGlobalPropertiesSet() { + BindingProperties bindingProperties = channelBindingServiceProperties.getBindingProperties(Source.OUTPUT); + Assertions.assertThat(bindingProperties.getContentType()).isEqualTo("application/json"); + Assertions.assertThat(bindingProperties.getDestination()).isEqualTo("ticktock"); + Assertions.assertThat(bindingProperties.getProducer().getRequiredGroups()).containsExactly("someGroup"); + Assertions.assertThat(bindingProperties.getProducer().getHeaderMode()).isEqualTo(HeaderMode.raw); + } + + @EnableBinding(Source.class) + @EnableAutoConfiguration + @Import(MockBinderRegistryConfiguration.class) + public static class TestSource { + + } +}