diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java index d26cae6ae..0927a78bd 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java @@ -43,10 +43,11 @@ abstract class AggregateApplication { public static final String OUTPUT_CHANNEL_NAME = "output"; - static ConfigurableApplicationContext createParentContext(String[] args, boolean selfContained) { + static ConfigurableApplicationContext createParentContext(Object[] sources, String[] args, boolean selfContained) { SpringApplicationBuilder aggregatorParentConfiguration = new SpringApplicationBuilder(); aggregatorParentConfiguration .sources(AggregatorParentConfiguration.class) + .sources(sources) .web(false) .headless(true) .properties("spring.jmx.default-domain=" 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 5baeba504..8e6e82782 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 @@ -21,12 +21,9 @@ import java.util.Arrays; import java.util.LinkedHashMap; import java.util.List; -import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.util.Assert; -import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; /** @@ -49,6 +46,10 @@ public class AggregateApplicationBuilder { private ConfigurableApplicationContext parentContext; + private List parentSources = new ArrayList<>(); + + private List parentArgs = new ArrayList<>(); + public AggregateApplicationBuilder(String... args) { this(new Object[]{ParentConfiguration.class}, args); } @@ -58,43 +59,28 @@ public class AggregateApplicationBuilder { } public AggregateApplicationBuilder(Object[] sources, String[] args) { - this(SpringApplication.run(addParentConfiguration(sources), args)); - } - - public AggregateApplicationBuilder(ConfigurableApplicationContext parentContext) { - this.parentContext = parentContext; + addParentSources(sources); + this.parentArgs.addAll(Arrays.asList(args)); } /** * Adding auto configuration classes to parent sources excluding the configuration * classes related to binder/binding. */ - private static Object[] addParentConfiguration(Object[] sources) { - Object[] parentSources; - if (!ObjectUtils.containsElement(sources, ParentConfiguration.class)) { - // add the ParentConfiguration first, so it can be overridden - List sourceList = new ArrayList<>(Arrays.asList(sources)); - sourceList.add(0, ParentConfiguration.class); - parentSources = sourceList.toArray(new Object[sourceList.size()]); + private void addParentSources(Object[] sources) { + if (!this.parentSources.contains(ParentConfiguration.class)) { + this.parentSources.add(ParentConfiguration.class); } - else { - parentSources = sources; - } - return parentSources; + this.parentSources.addAll(Arrays.asList(sources)); } - public AggregateApplicationBuilder parent(Object source, String... args) { return parent(new Object[]{source}, args); } public AggregateApplicationBuilder parent(Object[] sources, String[] args) { - return parent(SpringApplication.run(sources, args)); - } - - public AggregateApplicationBuilder parent(ConfigurableApplicationContext parentContext) { - Assert.isNull(this.parentContext, "A parent context has already been set"); - this.parentContext = parentContext; + addParentSources(sources); + this.parentArgs.addAll(Arrays.asList(args)); return this; } @@ -104,7 +90,8 @@ public class AggregateApplicationBuilder { return sourceConfigurer; } - public ConfigurableApplicationContext run(String[] parentArgs) { + public ConfigurableApplicationContext run(String[] parentArgsFromRun) { + this.parentArgs.addAll(Arrays.asList(parentArgsFromRun)); List> apps = new ArrayList>(); if (this.sourceConfigurer != null) { apps.add(sourceConfigurer); @@ -127,13 +114,8 @@ public class AggregateApplicationBuilder { } appsToEmbed.put(appToEmbed, appConfigurer.namespace); } - if (this.parentContext == null) { - this.parentContext = AggregateApplication.createParentContext(parentArgs, areAppsSelfContained()); - } - // make sure to use the parent context that has aggregator parent configuration - else if (this.parentContext.getBeansOfType(SharedChannelRegistry.class).isEmpty()) { - this.parentContext = AggregateApplication.createParentContext(this.parentContext, parentArgs, areAppsSelfContained()); - } + this.parentContext = AggregateApplication.createParentContext(this.parentSources.toArray(new Object[0]), + this.parentArgs.toArray(new String[0]), areAppsSelfContained()); SharedChannelRegistry sharedChannelRegistry = this.parentContext.getBean(SharedChannelRegistry.class); AggregateApplication.prepareSharedChannelRegistry(sharedChannelRegistry, appsToEmbed); for (int i = apps.size() - 1; i >= 0; i--) { 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 7865a740d..adf3755b7 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,6 +16,11 @@ package org.springframework.cloud.stream.aggregation; +import java.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; + +import org.junit.Assert; import org.junit.Test; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -27,12 +32,15 @@ import org.springframework.cloud.stream.messaging.Processor; import org.springframework.cloud.stream.messaging.Source; import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Configuration; import org.springframework.messaging.MessageChannel; +import org.springframework.util.ReflectionUtils; import static org.assertj.core.api.Assertions.assertThat; /** * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan */ public class ModuleAggregationTest { @@ -51,6 +59,32 @@ public class ModuleAggregationTest { aggregatedApplicationContext.close(); } + @Test + public void testParentArgsAndSources() { + List argsToVerify = new ArrayList<>(); + argsToVerify.add("--foo1=bar1"); + argsToVerify.add("--foo2=bar2"); + argsToVerify.add("--foo3=bar3"); + AggregateApplicationBuilder aggregateApplicationBuilder = + new AggregateApplicationBuilder(MockBinderRegistryConfiguration.class, + "--foo1=bar1"); + aggregateApplicationBuilder.parent(DummyConfig.class, "--foo2=bar2") + .from(TestSource.class) + .namespace("foo").to(TestProcessor.class).namespace("bar") + .run("--foo3=bar3"); + Field parentArgsField = ReflectionUtils.findField(AggregateApplicationBuilder.class,"parentArgs", List.class); + ReflectionUtils.makeAccessible(parentArgsField); + Field parentSourcesField = ReflectionUtils.findField(AggregateApplicationBuilder.class,"parentSources", List.class); + ReflectionUtils.makeAccessible(parentSourcesField); + String args = ReflectionUtils.getField(parentArgsField, aggregateApplicationBuilder).toString(); + Assert.assertEquals(args, argsToVerify.toString()); + List sources = ((List)ReflectionUtils.getField(parentSourcesField, aggregateApplicationBuilder)); + Assert.assertTrue(sources.size() == 3); + Assert.assertTrue(sources.contains(AggregateApplicationBuilder.ParentConfiguration.class)); + Assert.assertTrue(sources.contains(MockBinderRegistryConfiguration.class)); + Assert.assertTrue(sources.contains(DummyConfig.class)); + } + @Test public void testNamespaces() { ConfigurableApplicationContext aggregatedApplicationContext = @@ -87,4 +121,8 @@ public class ModuleAggregationTest { } + @Configuration + public static class DummyConfig { + + } }