Use 'args.aggregate' for aggregate parameters

spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncher.java

@@ -29,7 +29,6 @@ import java.util.regex.Pattern;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-
 import org.springframework.boot.loader.archive.Archive;
 import org.springframework.boot.loader.archive.JarFileArchive;
 import org.springframework.cloud.stream.module.resolver.ModuleResolver;
@@ -88,19 +87,20 @@ public class ModuleLauncher {
 	 * @param aggregate whether the modules should be aggregated at launch
 	 * @param parentArgs a list of arguments for the whole aggregate
 	 */
-	public void launch(List<ModuleLaunchRequest> moduleLaunchRequests, boolean aggregate, String parentArgs[]) {
+	public void launch(List<ModuleLaunchRequest> moduleLaunchRequests, boolean aggregate, Map<String,String> parentArgs) {
 		List<ModuleLaunchRequest> reversed = new ArrayList<>(moduleLaunchRequests);
 		Collections.reverse(reversed);
 		if (moduleLaunchRequests.size() == 1 || !aggregate) {
 			launchIndividualModules(moduleLaunchRequests);
 		}
 		else {
-			launchAggregatedModules(moduleLaunchRequests, parentArgs);
+			launchAggregatedModules(moduleLaunchRequests, toArgArray(parentArgs));
 		}
 	}
 
+	@SuppressWarnings("unchecked")
 	public void launch(List<ModuleLaunchRequest> moduleLaunchRequests, boolean aggregate) {
-		this.launch(moduleLaunchRequests, aggregate, new String[0]);
+		this.launch(moduleLaunchRequests, aggregate, Collections.EMPTY_MAP);
 	}
 
 	public void launch(List<ModuleLaunchRequest> moduleLaunchRequests) {
@@ -112,12 +112,17 @@ public class ModuleLauncher {
 	 * {@literal --foo=bar} form.
 	 */
 	private String[] toArgArray(Map<String, String> args) {
-		String[] result = new String[args.size()];
-		int i = 0;
-		for (Map.Entry<String, String> kv : args.entrySet()) {
-			result[i++] = String.format("--%s=%s", kv.getKey(), kv.getValue());
+		if (args != null) {
+			String[] result = new String[args.size()];
+			int i = 0;
+			for (Map.Entry<String, String> kv : args.entrySet()) {
+				result[i++] = String.format("--%s=%s", kv.getKey(), kv.getValue());
+			}
+			return result;
+		}
+		else {
+			return new String[0];
 		}
-		return result;
 	}
 
 	public void launchAggregatedModules(List<ModuleLaunchRequest> moduleLaunchRequests, final String[] parentArgs) {

spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherProperties.java

@@ -31,14 +31,25 @@ import org.springframework.core.env.Environment;
  * properties, environment variables, program arguments, <i>etc.</i> _):<ul>
  *     <li>{@literal modules = <list>}: an ordered list of maven coordinates of modules to launch</li>
  *     <li>{@literal args[<index>][<key>] = <value>}: key/value pairs that will become module arguments,
- *     where {@literal <index>} is the 0-based index of the module in the list above</li>
+ *     where {@literal <index>} is the 0-based index of the module in the list above and '*' can be used for passing
+ *     arguments to all modules</li>
+ *     <li>{@literal aggregate = true | false}</li>: whether multiple modules launched together should be aggregated,
+ *     case in which they will be launched as a single individual unit, and {@literal args['aggregate'][<key>] = <value>}
+ *     can be used for passing arguments to the aggregate;
  * </ul>
  *
- * As an example, this is how one would launch {@literal time --fixedDelay=4 | log} canonical example:
+ * As an example, this is how one would launch the {@literal time --fixedDelay=4 | log} canonical example:
  * <pre>
  *     modules = org.springframework.cloud.modules:time-source:1.0.0-SNAPSHOT,org.springframework.cloud.modules:log-sink:1.0.0-SNAPSHOT
  *     args.0.fixedDelay=4
  * </pre>
+ *
+ * And this is how one would launch the {@literal time --fixedDelay=4 | log} example as an aggregate:
+ * <pre>
+ *     modules = org.springframework.cloud.modules:time-source:1.0.0-SNAPSHOT,org.springframework.cloud.modules:log-sink:1.0.0-SNAPSHOT
+ *     args.0.fixedDelay=4
+ *     aggregate=true
+ * </pre>
  * </p>
  *
  * @author Ilayaperumal Gopinathan
@@ -64,10 +75,6 @@ public class ModuleLauncherProperties {
 	 */
 	private Map<String, Map<String, String>> args = new HashMap<>();
 
-	public void setModules(String[] modules) {
-		this.modules = modules;
-	}
-
 	public boolean isAggregate() {
 		return aggregate;
 	}
@@ -76,6 +83,10 @@ public class ModuleLauncherProperties {
 		this.aggregate = aggregate;
 	}
 
+	public void setModules(String[] modules) {
+		this.modules = modules;
+	}
+
 	@NotEmpty(message = "A list of modules must be specified.")
 	public String[] getModules() {
 		return modules;
@@ -88,5 +99,4 @@ public class ModuleLauncherProperties {
 	public Map<String, Map<String, String>> getArgs() {
 		return args;
 	}
-
 }

spring-cloud-stream-module-launcher/src/main/java/org/springframework/cloud/stream/module/launcher/ModuleLauncherRunner.java

@@ -45,6 +45,8 @@ public class ModuleLauncherRunner implements CommandLineRunner {
 
 	private final static String GLOBAL_ARGS_KEY = "*";
 
+	private final static String AGGREGATE_ARGS_KEY = "aggregate";
+
 	@Autowired
 	private ModuleLauncherProperties moduleLauncherProperties;
 
@@ -63,7 +65,7 @@ public class ModuleLauncherRunner implements CommandLineRunner {
 		}
 		this.moduleLauncher.launch(launchRequests,
 				moduleLauncherProperties.isAggregate(),
-				moduleLauncherProperties.isAggregate() ? extractAggregateProperties(args) : new String[0]);
+				moduleLauncherProperties.isAggregate() ? moduleLauncherProperties.getArgs().get(AGGREGATE_ARGS_KEY) : null);
 	}
 
 	private List<ModuleLaunchRequest> generateModuleLaunchRequests() {
@@ -91,14 +93,5 @@ public class ModuleLauncherRunner implements CommandLineRunner {
 		return requests;
 	}
 
-	private String[] extractAggregateProperties(String[] args) {
-		List<String> filteredProperties = new ArrayList<>();
-		for (String arg : args) {
-			if (arg.startsWith("--spring.")) {
-				filteredProperties.add(arg);
-			}
-		}
-		return filteredProperties.toArray(new String[filteredProperties.size()]);
-	}
 
 }