diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java index 09ea7047d..99102f15d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java @@ -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, String> appsToEmbed = new LinkedHashMap<>(); + LinkedHashMap 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 appConfigurerEntry : appConfigurers + .entrySet()) { + AppConfigurer appConfigurer = appConfigurerEntry.getKey(); + String namespace = appConfigurerEntry.getValue().toLowerCase(); + Set argsToUpdate = new LinkedHashSet<>(); + Set argKeys = new LinkedHashSet<>(); + final HashMap target = new HashMap<>(); + RelaxedDataBinder relaxedDataBinder = new RelaxedDataBinder(target, namespace); + relaxedDataBinder.bind(new PropertySourcesPropertyValues(propertySources)); + if (!target.isEmpty()) { + for (Map.Entry 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 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 { @@ -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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java index 8164f62f9..ca892484b 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java @@ -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) 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) 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) 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) 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) 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) 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 {