Pass aggregator parent sources/args appropriately

- The aggregator builder parent sources and cmdline args can be passed via
   - AggregateApplicationBuilder constructor
   - AggregateApplicationBuilder parent() builder method
and, the cmd line args can also be passed via
   - AggregateApplicationBuilder run() method

This PR supports passing all these options. The implementation sets `parentSources` and `parentArgs` by collecting the values from all the above options and use them when creating the parent context.

 - Add test case to verify these scenarios

This resolves #626

Make sure to handle self-contained apps parentContext

 - Always create parent context with the aggregatorParentConfiguration

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="

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<Object> parentSources = new ArrayList<>();
+
+	private List<String> 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<Object> 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<AppConfigurer<?>> apps = new ArrayList<AppConfigurer<?>>();
 		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--) {

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<String> 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<Object> sources = ((List<Object>)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 {
+
+	}
 }