Add AggregateApplicationBuilder
- Uses builder pattern to create aggregate application that supports specifying
- args
- profiles
- config name for each child app and parent aggregate app separately.
An example config would look like this:
```
@SpringBootApplication
public class DoubleApplication {
public static void main(String[] args) {
new AggregateApplicationBuilder().
from(SourceApplication.class).args("--fixedDelay=5000")
.to(SinkApplication.class).args("--debug=true").run("--spring.application.name=aggregate-test");
}
```
This resolves #362
This commit is contained in:
Ilayaperumal Gopinathan
committed by
Marius Bogoevici
Marius Bogoevici
parent
81e6769b74
commit
7d7fcc9184
@@ -17,20 +17,16 @@
|
||||
package org.springframework.cloud.stream.aggregate;
|
||||
|
||||
import org.springframework.boot.Banner.Mode;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.messaging.Processor;
|
||||
import org.springframework.cloud.stream.messaging.Sink;
|
||||
import org.springframework.cloud.stream.messaging.Source;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
|
||||
/**
|
||||
* Class that is responsible for embedding modules using shared channel registry.
|
||||
* Class that is responsible for embedding apps using shared channel registry.
|
||||
*
|
||||
* @author Marius Bogoevici
|
||||
* @author Ilayaperumal Gopinathan
|
||||
@@ -39,7 +35,7 @@ public class AggregateApplication {
|
||||
|
||||
private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";
|
||||
|
||||
private static final String CHANNEL_NAMESPACE_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
|
||||
public static final String CHANNEL_NAMESPACE_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
|
||||
|
||||
public static final String INPUT_CHANNEL_NAME = "input";
|
||||
|
||||
@@ -47,41 +43,41 @@ public class AggregateApplication {
|
||||
|
||||
/**
|
||||
* Supports the aggregation of {@link Source}, {@link Sink} and {@link Processor}
|
||||
* modules by instantiating and binding them directly
|
||||
* apps by instantiating and binding them directly
|
||||
*
|
||||
* @param parentArgs arguments for the parent (prefixed with '--')
|
||||
* @param modules a list module classes to be aggregated
|
||||
* @param moduleArgs arguments for the modules (prefixed with '--")
|
||||
* @param apps a list app classes to be aggregated
|
||||
* @param appArgs arguments for the apps (prefixed with '--")
|
||||
*
|
||||
* @return the resulting parent context for the aggregate
|
||||
*/
|
||||
public static ConfigurableApplicationContext run(Class<?>[] modules, String[] parentArgs, String[][] moduleArgs) {
|
||||
public static ConfigurableApplicationContext run(Class<?>[] apps, String[] parentArgs, String[][] appArgs) {
|
||||
ConfigurableApplicationContext parentContext = createParentContext(parentArgs != null ? parentArgs
|
||||
: new String[0]);
|
||||
runEmbedded(parentContext, modules, moduleArgs);
|
||||
runEmbedded(parentContext, apps, appArgs);
|
||||
return parentContext;
|
||||
}
|
||||
|
||||
public static ConfigurableApplicationContext run(Class<?>... modules) {
|
||||
return run(modules, null, null);
|
||||
public static ConfigurableApplicationContext run(Class<?>... apps) {
|
||||
return run(apps, null, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Embeds a group of modules into an existing parent context
|
||||
* Embeds a group of apps into an existing parent context
|
||||
*
|
||||
* @param parentContext the parent context
|
||||
* @param modules a list of classes, representing root context definitions for modules
|
||||
* @param args arguments for the modules
|
||||
* @param apps a list of classes, representing root context definitions for apps
|
||||
* @param args arguments for the apps
|
||||
*/
|
||||
public static void runEmbedded(ConfigurableApplicationContext parentContext,
|
||||
Class<?>[] modules, String[][] args) {
|
||||
Class<?>[] apps, String[][] args) {
|
||||
SharedChannelRegistry bean = parentContext.getBean(SharedChannelRegistry.class);
|
||||
prepareSharedChannelRegistry(bean, modules);
|
||||
prepareSharedChannelRegistry(bean, apps);
|
||||
// create child contexts first
|
||||
createChildContexts(parentContext, modules, args);
|
||||
createChildContexts(parentContext, apps, args);
|
||||
}
|
||||
|
||||
private static ConfigurableApplicationContext createParentContext(String[] args) {
|
||||
protected static ConfigurableApplicationContext createParentContext(String[] args) {
|
||||
SpringApplicationBuilder aggregatorParentConfiguration = new SpringApplicationBuilder();
|
||||
aggregatorParentConfiguration
|
||||
.sources(AggregatorParentConfiguration.class)
|
||||
@@ -93,59 +89,45 @@ public class AggregateApplication {
|
||||
}
|
||||
|
||||
private static void createChildContexts(ConfigurableApplicationContext parentContext,
|
||||
Class<?>[] modules, String args[][]) {
|
||||
for (int i = modules.length - 1; i >= 0; i--) {
|
||||
String moduleClassName = modules[i].getName();
|
||||
embedModule(parentContext, getNamespace(moduleClassName, i), modules[i])
|
||||
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])
|
||||
.run(args != null ? args[i] : new String[0]);
|
||||
}
|
||||
}
|
||||
|
||||
private static String getNamespace(String moduleClassName, int index) {
|
||||
return moduleClassName + "_" + index;
|
||||
protected static String getNamespace(String appClassName, int index) {
|
||||
return appClassName + "_" + index;
|
||||
}
|
||||
|
||||
private static SpringApplicationBuilder embedModule(
|
||||
protected static SpringApplicationBuilder embedApp(
|
||||
ConfigurableApplicationContext applicationContext, String namespace,
|
||||
Class<?> module) {
|
||||
return new SpringApplicationBuilder(module)
|
||||
Class<?> app) {
|
||||
return new SpringApplicationBuilder(app)
|
||||
.web(false)
|
||||
.bannerMode(Mode.OFF)
|
||||
.properties("spring.jmx.default-domain=" + module)
|
||||
.properties("spring.jmx.default-domain=" + app)
|
||||
.properties(CHANNEL_NAMESPACE_PROPERTY_NAME + "=" + namespace)
|
||||
.registerShutdownHook(false)
|
||||
.parent(applicationContext);
|
||||
}
|
||||
|
||||
private static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry, Class<?>[] modules) {
|
||||
protected static void prepareSharedChannelRegistry(SharedChannelRegistry sharedChannelRegistry, Class<?>[] apps) {
|
||||
SubscribableChannel sharedChannel = null;
|
||||
for (int i = 0; i < modules.length; i++) {
|
||||
Class<?> module = modules[i];
|
||||
String moduleClassName = module.getName();
|
||||
for (int i = 0; i < apps.length; i++) {
|
||||
Class<?> app = apps[i];
|
||||
String appClassName = app.getName();
|
||||
if (i > 0) {
|
||||
sharedChannelRegistry.register(getNamespace(moduleClassName, i)
|
||||
sharedChannelRegistry.register(getNamespace(appClassName, i)
|
||||
+ "." + INPUT_CHANNEL_NAME, sharedChannel);
|
||||
}
|
||||
sharedChannel = new DirectChannel();
|
||||
if (i < modules.length - 1) {
|
||||
sharedChannelRegistry.register(getNamespace(moduleClassName, i)
|
||||
if (i < apps.length - 1) {
|
||||
sharedChannelRegistry.register(getNamespace(appClassName, i)
|
||||
+ "." + OUTPUT_CHANNEL_NAME, sharedChannel);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Basic configuration for a parent
|
||||
*/
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding
|
||||
public static class AggregatorParentConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(SharedChannelRegistry.class)
|
||||
public SharedChannelRegistry sharedChannelRegistry() {
|
||||
return new SharedChannelRegistry();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -0,0 +1,250 @@
|
||||
/*
|
||||
* Copyright 2015-2016 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.cloud.stream.aggregate;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* Application builder for {@link AggregateApplication}.
|
||||
*
|
||||
* @author Dave Syer
|
||||
* @author Ilayaperumal Gopinathan
|
||||
*
|
||||
*/
|
||||
public class AggregateApplicationBuilder {
|
||||
|
||||
private SourceConfigurer sourceConfigurer;
|
||||
|
||||
private SinkConfigurer sinkConfigurer;
|
||||
|
||||
private List<ProcessorConfigurer> processorConfigurers = new ArrayList<>();
|
||||
|
||||
private AggregateApplicationBuilder applicationBuilder = this;
|
||||
|
||||
public SourceConfigurer from(Class<?> app) {
|
||||
SourceConfigurer sourceConfigurer = new SourceConfigurer(app);
|
||||
this.sourceConfigurer = sourceConfigurer;
|
||||
return sourceConfigurer;
|
||||
}
|
||||
|
||||
public void run(String[] parentArgs) {
|
||||
ConfigurableApplicationContext parentContext = AggregateApplication.createParentContext(parentArgs);
|
||||
SharedChannelRegistry sharedChannelRegistry = parentContext.getBean(SharedChannelRegistry.class);
|
||||
List<AppConfigurer> apps = new ArrayList<AppConfigurer>();
|
||||
if (this.sourceConfigurer != null) {
|
||||
apps.add(sourceConfigurer);
|
||||
}
|
||||
if (!processorConfigurers.isEmpty()) {
|
||||
for (ProcessorConfigurer processorConfigurer : processorConfigurers) {
|
||||
apps.add(processorConfigurer);
|
||||
}
|
||||
}
|
||||
if (this.sinkConfigurer != null) {
|
||||
apps.add(sinkConfigurer);
|
||||
}
|
||||
List<Class<?>> appsToEmbed = new ArrayList<>();
|
||||
for (int i = 0; i < apps.size(); i++) {
|
||||
appsToEmbed.add(apps.get(i).getApp());
|
||||
}
|
||||
AggregateApplication.prepareSharedChannelRegistry(sharedChannelRegistry, appsToEmbed.toArray(new Class<?>[0]));
|
||||
for (int i = apps.size() - 1; i >= 0; i--) {
|
||||
AppConfigurer appConfigurer = apps.get(i);
|
||||
appConfigurer.setParentContext(parentContext);
|
||||
appConfigurer.setNamespace(AggregateApplication.getNamespace(appConfigurer.getApp().getName(), i));
|
||||
appConfigurer.embed();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
public class SourceConfigurer extends AppConfigurer {
|
||||
|
||||
public SourceConfigurer(Class<?> app) {
|
||||
this.app = app;
|
||||
sourceConfigurer = this;
|
||||
}
|
||||
|
||||
public SourceConfigurer as(String... names) {
|
||||
this.names = names;
|
||||
return this;
|
||||
}
|
||||
|
||||
public SourceConfigurer args(String... args) {
|
||||
this.args = args;
|
||||
return this;
|
||||
}
|
||||
|
||||
public SourceConfigurer profiles(String... profiles) {
|
||||
this.profiles = profiles;
|
||||
return this;
|
||||
}
|
||||
|
||||
public SinkConfigurer to(Class<?> sink) {
|
||||
return new SinkConfigurer(sink);
|
||||
}
|
||||
|
||||
public ProcessorConfigurer via(Class<?> processor) {
|
||||
return new ProcessorConfigurer(processor);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public class SinkConfigurer extends AppConfigurer {
|
||||
|
||||
public SinkConfigurer(Class<?> app) {
|
||||
this.app = app;
|
||||
sinkConfigurer = this;
|
||||
}
|
||||
|
||||
public SinkConfigurer as(String... names) {
|
||||
this.names = names;
|
||||
return this;
|
||||
}
|
||||
|
||||
public SinkConfigurer args(String... args) {
|
||||
this.args = args;
|
||||
return this;
|
||||
}
|
||||
|
||||
public SinkConfigurer profiles(String... profiles) {
|
||||
this.profiles = profiles;
|
||||
return this;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public class ProcessorConfigurer extends AppConfigurer {
|
||||
|
||||
public ProcessorConfigurer(Class<?> app) {
|
||||
this.app = app;
|
||||
processorConfigurers.add(this);
|
||||
}
|
||||
|
||||
public ProcessorConfigurer as(String... names) {
|
||||
this.names = names;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ProcessorConfigurer args(String... args) {
|
||||
this.args = args;
|
||||
return this;
|
||||
}
|
||||
|
||||
public ProcessorConfigurer profiles(String... profiles) {
|
||||
this.profiles = profiles;
|
||||
return this;
|
||||
}
|
||||
|
||||
public SinkConfigurer to(Class<?> sink) {
|
||||
return new SinkConfigurer(sink);
|
||||
}
|
||||
|
||||
public ProcessorConfigurer via(Class<?> processor) {
|
||||
return new ProcessorConfigurer(processor);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private abstract class AppConfigurer {
|
||||
|
||||
Class<?> app;
|
||||
|
||||
String[] args;
|
||||
|
||||
String[] names = null;
|
||||
|
||||
String[] profiles = null;
|
||||
|
||||
ConfigurableApplicationContext parentContext;
|
||||
|
||||
String namespace;
|
||||
|
||||
Class<?> getApp() {
|
||||
return this.app;
|
||||
}
|
||||
|
||||
public void setParentContext(ConfigurableApplicationContext parentContext) {
|
||||
this.parentContext = parentContext;
|
||||
}
|
||||
|
||||
public void setNamespace(String namespace) {
|
||||
this.namespace = namespace;
|
||||
}
|
||||
|
||||
public void run(String... args) {
|
||||
applicationBuilder.run(args);
|
||||
}
|
||||
|
||||
void embed() {
|
||||
childContext(this.app, this.parentContext, this.namespace).args(this.args).config(this.names).profiles(this.profiles).run();
|
||||
}
|
||||
}
|
||||
|
||||
private ChildContextBuilder childContext(Class<?> app, ConfigurableApplicationContext parentContext, String namespace) {
|
||||
return new ChildContextBuilder(AggregateApplication.embedApp(parentContext, namespace, app));
|
||||
}
|
||||
|
||||
private class ChildContextBuilder {
|
||||
|
||||
private SpringApplicationBuilder builder;
|
||||
|
||||
private String configName;
|
||||
|
||||
private String[] args;
|
||||
|
||||
public ChildContextBuilder(SpringApplicationBuilder builder) {
|
||||
this.builder = builder;
|
||||
}
|
||||
|
||||
public ChildContextBuilder profiles(String... profiles) {
|
||||
if (profiles != null) {
|
||||
this.builder.profiles(profiles);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChildContextBuilder config(String... configs) {
|
||||
if (configs != null) {
|
||||
this.configName = StringUtils.arrayToCommaDelimitedString(configs);
|
||||
}
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChildContextBuilder args(String... args) {
|
||||
this.args = args;
|
||||
return this;
|
||||
}
|
||||
|
||||
public void run() {
|
||||
List<String> args = new ArrayList<String>();
|
||||
if (this.args != null) {
|
||||
args.addAll(Arrays.asList(this.args));
|
||||
}
|
||||
if (this.configName != null) {
|
||||
args.add("--spring.config.name=" + this.configName);
|
||||
}
|
||||
this.builder.run(args.toArray(new String[0]));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -82,7 +82,9 @@ public class AggregateBuilder implements ApplicationContextAware {
|
||||
public class SourceConfigurer {
|
||||
|
||||
private Class<?> module;
|
||||
|
||||
private String[] names = null;
|
||||
|
||||
private String[] profiles = null;
|
||||
|
||||
public SourceConfigurer(Class<?> module) {
|
||||
@@ -111,7 +113,7 @@ public class AggregateBuilder implements ApplicationContextAware {
|
||||
|
||||
private void build() {
|
||||
childContext(this.module).config(this.names).profiles(this.profiles)
|
||||
.output(channelName()).build();
|
||||
.output(channelName()).build();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -119,7 +121,9 @@ public class AggregateBuilder implements ApplicationContextAware {
|
||||
public class SinkConfigurer {
|
||||
|
||||
private Class<?> module;
|
||||
|
||||
private String[] names = null;
|
||||
|
||||
private String[] profiles = null;
|
||||
|
||||
public SinkConfigurer profiles(String... profiles) {
|
||||
@@ -139,7 +143,7 @@ public class AggregateBuilder implements ApplicationContextAware {
|
||||
|
||||
void build() {
|
||||
childContext(this.module).config(this.names).profiles(this.profiles)
|
||||
.input(channelName()).build();
|
||||
.input(channelName()).build();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -147,7 +151,9 @@ public class AggregateBuilder implements ApplicationContextAware {
|
||||
public class ProcessorConfigurer {
|
||||
|
||||
private Class<?> module;
|
||||
|
||||
private String[] names = null;
|
||||
|
||||
private String[] profiles = null;
|
||||
|
||||
public ProcessorConfigurer(Class<?> module) {
|
||||
@@ -176,7 +182,7 @@ public class AggregateBuilder implements ApplicationContextAware {
|
||||
|
||||
private void build() {
|
||||
childContext(this.module).config(this.names).profiles(this.profiles)
|
||||
.input(incrementChannelName()).output(channelName()).build();
|
||||
.input(incrementChannelName()).output(channelName()).build();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -191,8 +197,11 @@ public class AggregateBuilder implements ApplicationContextAware {
|
||||
private class ChildContextBuilder {
|
||||
|
||||
private SpringApplicationBuilder builder;
|
||||
|
||||
private String configName;
|
||||
|
||||
private String input;
|
||||
|
||||
private String output;
|
||||
|
||||
public ChildContextBuilder(SpringApplicationBuilder builder) {
|
||||
@@ -240,7 +249,7 @@ public class AggregateBuilder implements ApplicationContextAware {
|
||||
}
|
||||
|
||||
private class BeanPostProcessorInitializer implements
|
||||
ApplicationContextInitializer<ConfigurableApplicationContext> {
|
||||
ApplicationContextInitializer<ConfigurableApplicationContext> {
|
||||
|
||||
@Override
|
||||
public void initialize(ConfigurableApplicationContext applicationContext) {
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
/*
|
||||
* Copyright 2015-2016 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.springframework.cloud.stream.aggregate;
|
||||
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
|
||||
/**
|
||||
* Basic configuration for an aggregator application parent.
|
||||
*
|
||||
* @author Marius Bogoevici
|
||||
*/
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding
|
||||
public class AggregatorParentConfiguration {
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(SharedChannelRegistry.class)
|
||||
public SharedChannelRegistry sharedChannelRegistry() {
|
||||
return new SharedChannelRegistry();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user