Fixes #428 - Enabled namespace propagation at AggregateApplication run.
Updated AggregationApplication* to honor app specific namespace.
This commit is contained in:
committed by
Ilayaperumal Gopinathan
parent
12bd4effa1
commit
99f978a262
@@ -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 ++;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user