Support namespace based prefixed properties

- In AggregateApplication, support specifying commandline and environment properties based on namespace for each containing applications using namespace as prefix
  - As usual, command line properties override the environment as well as the properties specified inside the application
This commit is contained in:
Ilayaperumal Gopinathan
2016-09-07 14:31:22 +05:30
committed by Marius Bogoevici
parent 8c33609f2b
commit 6fc6445c4f
2 changed files with 264 additions and 27 deletions

View File

@@ -18,17 +18,26 @@ package org.springframework.cloud.stream.aggregate;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.springframework.boot.actuate.endpoint.MetricReaderPublicMetrics;
import org.springframework.boot.actuate.endpoint.MetricsEndpoint;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.bind.PropertySourcesPropertyValues;
import org.springframework.boot.bind.RelaxedDataBinder;
import org.springframework.boot.bind.RelaxedNames;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.core.env.PropertySources;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.util.StringUtils;
@@ -65,7 +74,7 @@ public class AggregateApplicationBuilder {
}
public AggregateApplicationBuilder(Object source, String... args) {
this(new Object[]{ source }, args);
this(new Object[] { source }, args);
}
public AggregateApplicationBuilder(Object[] sources, String[] args) {
@@ -85,7 +94,7 @@ public class AggregateApplicationBuilder {
}
public AggregateApplicationBuilder parent(Object source, String... args) {
return parent(new Object[]{ source }, args);
return parent(new Object[] { source }, args);
}
public AggregateApplicationBuilder parent(Object[] sources, String[] args) {
@@ -139,6 +148,7 @@ public class AggregateApplicationBuilder {
apps.add(sinkConfigurer);
}
LinkedHashMap<Class<?>, String> appsToEmbed = new LinkedHashMap<>();
LinkedHashMap<AppConfigurer, String> appConfigurers = new LinkedHashMap<>();
for (int i = 0; i < apps.size(); i++) {
AppConfigurer<?> appConfigurer = apps.get(i);
Class<?> appToEmbed = appConfigurer.getApp();
@@ -148,11 +158,53 @@ public class AggregateApplicationBuilder {
i);
}
appsToEmbed.put(appToEmbed, appConfigurer.namespace);
appConfigurers.put(appConfigurer, appConfigurer.namespace);
}
this.parentContext = AggregateApplication.createParentContext(this.parentSources.toArray(new Object[0]),
this.parentArgs.toArray(new String[0]), selfContained(), this.webEnvironment, this.headless);
SharedChannelRegistry sharedChannelRegistry = this.parentContext.getBean(SharedChannelRegistry.class);
AggregateApplication.prepareSharedChannelRegistry(sharedChannelRegistry, appsToEmbed);
PropertySources propertySources = this.parentContext.getEnvironment()
.getPropertySources();
for (Map.Entry<AppConfigurer, String> appConfigurerEntry : appConfigurers
.entrySet()) {
AppConfigurer appConfigurer = appConfigurerEntry.getKey();
String namespace = appConfigurerEntry.getValue().toLowerCase();
Set<String> argsToUpdate = new LinkedHashSet<>();
Set<String> argKeys = new LinkedHashSet<>();
final HashMap<String, String> target = new HashMap<>();
RelaxedDataBinder relaxedDataBinder = new RelaxedDataBinder(target, namespace);
relaxedDataBinder.bind(new PropertySourcesPropertyValues(propertySources));
if (!target.isEmpty()) {
for (Map.Entry<String, String> entry : target.entrySet()) {
// only update the values with the highest precedence level.
if (!relaxedNameKeyExists(entry.getKey(), argKeys)) {
String key = entry.getKey();
// in case of environment variables pass the lower-case property key
// as we pass the properties as command line properties
if (key.contains("_")) {
key = key.replace("_", "-").toLowerCase();
}
argKeys.add(key);
argsToUpdate.add("--" + key + "=" + entry.getValue());
}
}
}
// Add the args that are set at the application level if they weren't
// overridden above from other property sources.
if (appConfigurer.getArgs() != null) {
for (String arg: appConfigurer.getArgs()) {
// use the key part left to the assignment and trimming the prefix `--`
String key = arg.substring(0, arg.indexOf("=")).substring(2);
if (!relaxedNameKeyExists(key, argKeys)) {
argsToUpdate.add(arg);
}
}
}
if (!argsToUpdate.isEmpty()) {
appConfigurer.args(argsToUpdate.toArray(new String[0]));
}
}
for (int i = apps.size() - 1; i >= 0; i--) {
AppConfigurer<?> appConfigurer = apps.get(i);
appConfigurer.embed();
@@ -164,12 +216,21 @@ public class AggregateApplicationBuilder {
return (this.sourceConfigurer != null) && (this.sinkConfigurer != null);
}
private ChildContextBuilder childContext(Class<?> app,
ConfigurableApplicationContext parentContext, String namespace) {
return new ChildContextBuilder(
AggregateApplication.embedApp(parentContext, namespace, app));
private boolean relaxedNameKeyExists(String key, Collection<String> collection) {
RelaxedNames relaxedNames = new RelaxedNames(key);
for (String name : relaxedNames) {
if (collection.contains(name)) {
return true;
}
}
return false;
}
private ChildContextBuilder childContext(Class<?> app,
ConfigurableApplicationContext parentContext, String namespace) {
return new ChildContextBuilder(AggregateApplication.embedApp(parentContext,
namespace, app));
}
public class SourceConfigurer extends AppConfigurer<SourceConfigurer> {
@@ -270,6 +331,14 @@ public class AggregateApplicationBuilder {
new NamespaceAwareSpringIntegrationMetricReader(this.namespace, childContext.getBean(
IntegrationMBeanExporter.class))));
}
public String[] getArgs() {
return this.args;
}
public String getNamespace() {
return this.namespace;
}
}
private final class ChildContextBuilder {

View File

@@ -16,14 +16,20 @@
package org.springframework.cloud.stream.aggregation;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.junit.After;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder;
import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder.SourceConfigurer;
import org.springframework.cloud.stream.aggregate.SharedChannelRegistry;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binding.BindableChannelFactory;
@@ -34,24 +40,34 @@ import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.MessageChannel;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
*/
public class ModuleAggregationTest {
private ConfigurableApplicationContext aggregatedApplicationContext;
@After
public void closeContext() {
System.clearProperty("a.foo-value");
System.clearProperty("c.fooValue");
System.clearProperty("a_FOO_VALUE");
System.clearProperty("C_FOO_VALUE");
if (aggregatedApplicationContext != null) {
aggregatedApplicationContext.close();
}
}
@Test
public void testModuleAggregation() {
ConfigurableApplicationContext aggregatedApplicationContext =
new AggregateApplicationBuilder(MockBinderRegistryConfiguration.class,
"--server.port=0")
.from(TestSource.class)
.to(TestProcessor.class)
.run();
SharedChannelRegistry sharedChannelRegistry = aggregatedApplicationContext.getBean(SharedChannelRegistry.class);
BindableChannelFactory channelFactory = aggregatedApplicationContext.getBean(BindableChannelFactory.class);
aggregatedApplicationContext = new AggregateApplicationBuilder(
MockBinderRegistryConfiguration.class, "--server.port=0")
.from(TestSource.class).to(TestProcessor.class).run();
SharedChannelRegistry sharedChannelRegistry = aggregatedApplicationContext
.getBean(SharedChannelRegistry.class);
BindableChannelFactory channelFactory = aggregatedApplicationContext
.getBean(BindableChannelFactory.class);
assertThat(channelFactory).isNotNull();
assertThat(sharedChannelRegistry.getAll().keySet()).hasSize(2);
aggregatedApplicationContext.close();
@@ -81,18 +97,171 @@ public class ModuleAggregationTest {
context.close();
}
@Test
public void testNamespacePrefixesFromCmdLine() {
AggregateApplicationBuilder aggregateApplicationBuilder = new AggregateApplicationBuilder(
MockBinderRegistryConfiguration.class);
aggregatedApplicationContext = aggregateApplicationBuilder.parent(DummyConfig.class).from(TestSource.class)
.namespace("a").via(TestProcessor.class).namespace("b")
.via(TestProcessor.class).namespace("c")
.run("--a.foo1=bar1", "--b.foo1=bar2", "--c.foo1=bar3");
DirectFieldAccessor aggregateApplicationBuilderAccessor = new DirectFieldAccessor(aggregateApplicationBuilder);
assertTrue(Arrays.equals(((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")).getArgs(),
new String[] {"--foo1=bar1"}));
for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : ((List<AggregateApplicationBuilder.ProcessorConfigurer>) aggregateApplicationBuilderAccessor.getPropertyValue("processorConfigurers"))) {
if (processorConfigurer.getNamespace().equals("b")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--foo1=bar2" }));
}
if (processorConfigurer.getNamespace().equals("c")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--foo1=bar3" }));
}
}
aggregatedApplicationContext.close();
}
@Test
public void testNamespacePrefixesFromCmdLineVsArgs() {
AggregateApplicationBuilder aggregateApplicationBuilder = new AggregateApplicationBuilder(
MockBinderRegistryConfiguration.class);
aggregatedApplicationContext = aggregateApplicationBuilder.parent(DummyConfig.class).from(TestSource.class)
.namespace("a").args("--fooValue=bar")
.via(TestProcessor.class).namespace("b").args("--foo1=argbarb")
.via(TestProcessor.class).namespace("c")
.run("--a.fooValue=bara", "--c.foo1=barc");
DirectFieldAccessor aggregateApplicationBuilderAccessor = new DirectFieldAccessor(aggregateApplicationBuilder);
assertTrue(Arrays.equals(((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")).getArgs(),
new String[] {"--fooValue=bara"}));
for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : ((List<AggregateApplicationBuilder.ProcessorConfigurer>) aggregateApplicationBuilderAccessor.getPropertyValue("processorConfigurers"))) {
if (processorConfigurer.getNamespace().equals("b")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--foo1=argbarb" }));
}
if (processorConfigurer.getNamespace().equals("c")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--foo1=barc" }));
}
}
aggregatedApplicationContext.close();
}
@Test
public void testNamespacePrefixesFromCmdLineWithRelaxedNames() {
AggregateApplicationBuilder aggregateApplicationBuilder = new AggregateApplicationBuilder(
MockBinderRegistryConfiguration.class);
aggregatedApplicationContext = aggregateApplicationBuilder.parent(DummyConfig.class).from(TestSource.class)
.namespace("a").args("--foo-value=bar")
.via(TestProcessor.class).namespace("b").args("--fooValue=argbarb")
.via(TestProcessor.class).namespace("c")
.run("--a.fooValue=bara", "--b.foo-value=barb", "--c.foo1=barc");
DirectFieldAccessor aggregateApplicationBuilderAccessor = new DirectFieldAccessor(aggregateApplicationBuilder);
assertTrue(Arrays.equals(((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")).getArgs(),
new String[] {"--fooValue=bara"}));
for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : ((List<AggregateApplicationBuilder.ProcessorConfigurer>) aggregateApplicationBuilderAccessor.getPropertyValue("processorConfigurers"))) {
if (processorConfigurer.getNamespace().equals("b")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--foo-value=barb" }));
}
if (processorConfigurer.getNamespace().equals("c")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--foo1=barc" }));
}
}
aggregatedApplicationContext.close();
}
@Test
public void testNamespacePrefixesFromCmdLineWithRelaxedNamesAndMorePropertySources() {
AggregateApplicationBuilder aggregateApplicationBuilder = new AggregateApplicationBuilder(
MockBinderRegistryConfiguration.class);
System.setProperty("a.foo-value", "sysbara");
System.setProperty("c.fooValue", "sysbarc");
aggregatedApplicationContext = aggregateApplicationBuilder.parent(DummyConfig.class).from(TestSource.class)
.namespace("a").args("--foo-value=bar")
.via(TestProcessor.class).namespace("b").args("--fooValue=argbarb")
.via(TestProcessor.class).namespace("c").args("--foo-value=argbarc")
.run("--a.fooValue=bara");
DirectFieldAccessor aggregateApplicationBuilderAccessor = new DirectFieldAccessor(aggregateApplicationBuilder);
assertTrue(Arrays.equals(((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")).getArgs(),
new String[] {"--fooValue=bara"}));
for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : ((List<AggregateApplicationBuilder.ProcessorConfigurer>) aggregateApplicationBuilderAccessor.getPropertyValue("processorConfigurers"))) {
if (processorConfigurer.getNamespace().equals("b")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--fooValue=argbarb" }));
}
if (processorConfigurer.getNamespace().equals("c")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--fooValue=sysbarc" }));
}
}
aggregatedApplicationContext.close();
}
@Test
public void testNamespacePrefixesWithoutCmdLinePropertySource() {
AggregateApplicationBuilder aggregateApplicationBuilder = new AggregateApplicationBuilder(
MockBinderRegistryConfiguration.class);
System.setProperty("a.foo-value", "sysbara");
System.setProperty("c.fooValue", "sysbarc");
aggregatedApplicationContext = aggregateApplicationBuilder.parent(DummyConfig.class).from(TestSource.class)
.namespace("a").args("--foo-value=bar")
.via(TestProcessor.class).namespace("b").args("--fooValue=argbarb")
.via(TestProcessor.class).namespace("c").args("--foo-value=argbarc")
.run();
DirectFieldAccessor aggregateApplicationBuilderAccessor = new DirectFieldAccessor(aggregateApplicationBuilder);
assertTrue(Arrays.equals(((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")).getArgs(),
new String[] {"--foo-value=sysbara"}));
for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : ((List<AggregateApplicationBuilder.ProcessorConfigurer>) aggregateApplicationBuilderAccessor.getPropertyValue("processorConfigurers"))) {
if (processorConfigurer.getNamespace().equals("b")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--fooValue=argbarb" }));
}
if (processorConfigurer.getNamespace().equals("c")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--fooValue=sysbarc" }));
}
}
aggregatedApplicationContext.close();
}
@Test
public void testNamespacePrefixesWithCAPSProperties() {
AggregateApplicationBuilder aggregateApplicationBuilder = new AggregateApplicationBuilder(
MockBinderRegistryConfiguration.class);
System.setProperty("a_FOO_VALUE", "sysbara");
System.setProperty("C_FOO_VALUE", "sysbarc");
aggregatedApplicationContext = aggregateApplicationBuilder.parent(DummyConfig.class).from(TestSource.class)
.namespace("a").args("--foo-value=bar")
.via(TestProcessor.class).namespace("b").args("--fooValue=argbarb")
.via(TestProcessor.class).namespace("c").args("--foo-value=argbarc")
.run("--a.fooValue=highest");
DirectFieldAccessor aggregateApplicationBuilderAccessor = new DirectFieldAccessor(aggregateApplicationBuilder);
assertTrue(Arrays.equals(((SourceConfigurer) aggregateApplicationBuilderAccessor.getPropertyValue("sourceConfigurer")).getArgs(),
new String[] {"--fooValue=highest"}));
for (AggregateApplicationBuilder.ProcessorConfigurer processorConfigurer : ((List<AggregateApplicationBuilder.ProcessorConfigurer>) aggregateApplicationBuilderAccessor.getPropertyValue("processorConfigurers"))) {
if (processorConfigurer.getNamespace().equals("b")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--fooValue=argbarb" }));
}
if (processorConfigurer.getNamespace().equals("c")) {
assertTrue(Arrays.equals(processorConfigurer.getArgs(),
new String[] { "--foo-value=sysbarc" }));
}
}
aggregatedApplicationContext.close();
}
@Test
public void testNamespaces() {
ConfigurableApplicationContext aggregatedApplicationContext =
new AggregateApplicationBuilder(MockBinderRegistryConfiguration.class,
"--server.port=0")
.from(TestSource.class)
.namespace("foo").to(TestProcessor.class).namespace("bar")
.run();
SharedChannelRegistry sharedChannelRegistry
= aggregatedApplicationContext.getBean(SharedChannelRegistry.class);
BindableChannelFactory channelFactory
= aggregatedApplicationContext.getBean(BindableChannelFactory.class);
aggregatedApplicationContext = new AggregateApplicationBuilder(
MockBinderRegistryConfiguration.class, "--server.port=0")
.from(TestSource.class).namespace("foo").to(TestProcessor.class)
.namespace("bar").run();
SharedChannelRegistry sharedChannelRegistry = aggregatedApplicationContext
.getBean(SharedChannelRegistry.class);
BindableChannelFactory channelFactory = aggregatedApplicationContext
.getBean(BindableChannelFactory.class);
Object fooOutput = sharedChannelRegistry.get("foo.output");
assertThat(fooOutput).isNotNull();
assertThat(fooOutput).isInstanceOf(MessageChannel.class);
@@ -104,7 +273,6 @@ public class ModuleAggregationTest {
aggregatedApplicationContext.close();
}
@EnableBinding(Source.class)
@EnableAutoConfiguration
public static class TestSource {