SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add support for defaults at binding level

Browse Source 
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
This commit is contained in:
Marius Bogoevici
2016-12-01 18:33:22 -05:00
committed by Ilayaperumal Gopinathan
parent 6c861135a3
commit b3a0f2c2e1
4 changed files with 156 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc
View File

@@ -885,17 +885,21 @@ Default: false.
Binding properties are supplied using the format `spring.cloud.stream.bindings.<channelName>.<property>=<value>`.
The `<channelName>` 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.<property>=<value>`.
In what follows, we indicate where we have omitted the `spring.cloud.stream.bindings.<channelName>.` 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.<channelName>.`.
The following binding properties are available for both input and output bindings and must be prefixed with `spring.cloud.stream.bindings.<channelName>.`, 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.<channelName>.consumer.`.
The following binding properties are available for input bindings only and must be prefixed with `spring.cloud.stream.bindings.<channelName>.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.<channelName>.producer.`.
The following binding properties are available for output bindings only and must be prefixed with `spring.cloud.stream.bindings.<channelName>.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.

35
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java
View File

@@ -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<String, BinderProperties> 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<String, BindingProperties> 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, BindingProperties>(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();
}
}

75
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnvironmentEntryInitializingTreeMap.java Normal file
View File

@@ -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<T> extends AbstractMap<String, T> {
private final ConfigurableApplicationContext applicationContext;
private final Class<T> entryClass;
private final String defaultsPrefix;
private final Map<String, T> 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<T> entryClass,
String defaultsPrefix, Map<String, T> 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<Entry<String, T>> entrySet() {
return delegate.entrySet();
}
}

65
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java Normal file
View File

@@ -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 {
}
}