From 99f978a262a500c16e8c12c7d3bc9d775ff431c5 Mon Sep 17 00:00:00 2001
From: Venil Noronha <venil.noronha@gmail.com>
Date: Fri, 25 Mar 2016 22:11:41 +0530
Subject: [PATCH] Fixes #428 - Enabled namespace propagation at
 AggregateApplication run.

Updated AggregationApplication* to honor app specific namespace.
---
 .../aggregate/AggregateApplication.java       | 33 +++++++++++++++----
 .../AggregateApplicationBuilder.java          | 13 +++++---
 2 files changed, 34 insertions(+), 12 deletions(-)

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 6ed727729..97bc6a07c 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
@@ -16,6 +16,9 @@
 
 package org.springframework.cloud.stream.aggregate;
 
+import java.util.LinkedHashMap;
+import java.util.Map.Entry;
+
 import org.springframework.boot.Banner.Mode;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.stream.messaging.Processor;
@@ -30,6 +33,7 @@ import org.springframework.messaging.SubscribableChannel;
  *
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
+ * @author Venil Noronha
  */
 public class AggregateApplication {
 
@@ -92,12 +96,15 @@ public class AggregateApplication {
 			Class<?>[] apps, String args[][]) {
 		for (int i = apps.length - 1; i >= 0; i--) {
 			String appClassName = apps[i].getName();
-			embedApp(parentContext, getNamespace(appClassName, i), apps[i])
+			embedApp(parentContext, getNamespace(null, appClassName, i), apps[i])
 					.run(args != null ? args[i] : new String[0]);
 		}
 	}
 
-	protected static String getNamespace(String appClassName, int index) {
+	protected static String getNamespace(String namespace, String appClassName, int index) {
+		if (namespace != null) {
+			return namespace;
+		}
 		return appClassName + "_" + index;
 	}
 
@@ -114,19 +121,31 @@ public class AggregateApplication {
 	}
 
 	protected static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry, Class<?>[] apps) {
+		LinkedHashMap<Class<?>, String> appsToRegister = new LinkedHashMap<>();
+		for (Class<?> app : apps) {
+			appsToRegister.put(app, null);
+		}
+		prepareSharedChannelRegistry(sharedChannelRegistry, appsToRegister);
+	}
+
+	protected static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry,
+			LinkedHashMap<Class<?>, String> appsWithNamespace) {
+		int i = 0;
 		SubscribableChannel sharedChannel = null;
-		for (int i = 0; i < apps.length; i++) {
-			Class<?> app = apps[i];
+		for (Entry<Class<?>, String> appEntry : appsWithNamespace.entrySet()) {
+			Class<?> app = appEntry.getKey();
+			String namespace = appEntry.getValue();
 			String appClassName = app.getName();
 			if (i > 0) {
-				sharedChannelRegistry.register(getNamespace(appClassName, i)
+				sharedChannelRegistry.register(getNamespace(namespace, appClassName, i)
 						+ "." + INPUT_CHANNEL_NAME, sharedChannel);
 			}
 			sharedChannel = new DirectChannel();
-			if (i < apps.length - 1) {
-				sharedChannelRegistry.register(getNamespace(appClassName, i)
+			if (i < appsWithNamespace.size() - 1) {
+				sharedChannelRegistry.register(getNamespace(namespace, appClassName, i)
 						+ "." + OUTPUT_CHANNEL_NAME, sharedChannel);
 			}
+			i ++;
 		}
 	}
 
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 3478d5bcc..bfb0f06f3 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,6 +18,7 @@ package org.springframework.cloud.stream.aggregate;
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.LinkedHashMap;
 import java.util.List;
 
 import org.springframework.boot.SpringApplication;
@@ -33,6 +34,7 @@ import org.springframework.util.StringUtils;
  * @author Dave Syer
  * @author Ilayaperumal Gopinathan
  * @author Marius Bogoevici
+ * @author Venil Noronha
  */
 public class AggregateApplicationBuilder {
 
@@ -115,16 +117,17 @@ public class AggregateApplicationBuilder {
 		if (this.sinkConfigurer != null) {
 			apps.add(sinkConfigurer);
 		}
-		List<Class<?>> appsToEmbed = new ArrayList<>();
+		LinkedHashMap<Class<?>, String> appsToEmbed = new LinkedHashMap<>();
 		for (int i = 0; i < apps.size(); i++) {
-			appsToEmbed.add(apps.get(i).getApp());
+			AppConfigurer<?> appConfigurer = apps.get(i);
+			Class<?> appToEmbed = appConfigurer.getApp();
+			appsToEmbed.put(appToEmbed, appConfigurer.namespace);
 		}
-		AggregateApplication.prepareSharedChannelRegistry(sharedChannelRegistry,
-				appsToEmbed.toArray(new Class<?>[0]));
+		AggregateApplication.prepareSharedChannelRegistry(sharedChannelRegistry, appsToEmbed);
 		for (int i = apps.size() - 1; i >= 0; i--) {
 			AppConfigurer<?> appConfigurer = apps.get(i);
 			appConfigurer.namespace(AggregateApplication
-					.getNamespace(appConfigurer.getApp().getName(), i));
+					.getNamespace(appConfigurer.namespace, appConfigurer.getApp().getName(), i));
 			appConfigurer.embed();
 		}
 		return parentContext;