From 7d7fcc9184b95455090f0b0accf53db3354a323c Mon Sep 17 00:00:00 2001 From: Ilayaperumal Gopinathan Date: Mon, 22 Feb 2016 16:46:52 +0530 Subject: [PATCH] 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 --- .../aggregate/AggregateApplication.java | 84 +++--- .../AggregateApplicationBuilder.java | 250 ++++++++++++++++++ .../stream/aggregate/AggregateBuilder.java | 17 +- .../AggregatorParentConfiguration.java | 37 +++ 4 files changed, 333 insertions(+), 55 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregatorParentConfiguration.java 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 39247228a..6ed727729 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 @@ -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(); - } - } - } 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 new file mode 100644 index 000000000..c038fbd8d --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java @@ -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 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 apps = new ArrayList(); + 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> 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 args = new ArrayList(); + 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])); + } + + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java index f895fa35c..4a3a2afb1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java @@ -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 { + ApplicationContextInitializer { @Override public void initialize(ConfigurableApplicationContext applicationContext) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregatorParentConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregatorParentConfiguration.java new file mode 100644 index 000000000..05ef1cbe0 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregatorParentConfiguration.java @@ -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(); + } +}