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 deleted file mode 100644 index 4a3a2afb1..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateBuilder.java +++ /dev/null @@ -1,265 +0,0 @@ -/* - * Copyright 2015 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.HashSet; -import java.util.List; - -import org.springframework.beans.BeansException; -import org.springframework.boot.Banner.Mode; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.ApplicationContextInitializer; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Configuration; -import org.springframework.util.StringUtils; - -/** - * @author Dave Syer - * - */ -@ConfigurationProperties("spring.cloud.streams") -public class AggregateBuilder implements ApplicationContextAware { - - private ConfigurableApplicationContext parent; - - private int index = 0; - - private String streamName = "stream"; - - private HashSet sinks = new HashSet(); - - public String getName() { - return this.streamName; - } - - public void setName(String name) { - this.streamName = name; - } - - @Override - public void setApplicationContext(ApplicationContext applicationContext) - throws BeansException { - this.parent = (ConfigurableApplicationContext) applicationContext; - } - - public void build() { - for (SinkConfigurer sink : this.sinks) { - sink.build(); - } - } - - public SourceConfigurer from(Class module) { - return new SourceConfigurer(module); - } - - private String channelName() { - return this.streamName + "." + this.index; - } - - private String incrementChannelName() { - return this.streamName + "." + (this.index++); - } - - public class SourceConfigurer { - - private Class module; - - private String[] names = null; - - private String[] profiles = null; - - public SourceConfigurer(Class module) { - this.module = module; - } - - public SourceConfigurer as(String... names) { - this.names = names; - return this; - } - - public SourceConfigurer profiles(String... profiles) { - this.profiles = profiles; - return this; - } - - public SinkConfigurer to(Class sink) { - build(); - return new SinkConfigurer(sink); - } - - public ProcessorConfigurer via(Class processor) { - build(); - return new ProcessorConfigurer(processor); - } - - private void build() { - childContext(this.module).config(this.names).profiles(this.profiles) - .output(channelName()).build(); - } - - } - - public class SinkConfigurer { - - private Class module; - - private String[] names = null; - - private String[] profiles = null; - - public SinkConfigurer profiles(String... profiles) { - this.profiles = profiles; - return this; - } - - public SinkConfigurer(Class module) { - AggregateBuilder.this.sinks.add(this); - this.module = module; - } - - public SinkConfigurer as(String... names) { - this.names = names; - return this; - } - - void build() { - childContext(this.module).config(this.names).profiles(this.profiles) - .input(channelName()).build(); - } - - } - - public class ProcessorConfigurer { - - private Class module; - - private String[] names = null; - - private String[] profiles = null; - - public ProcessorConfigurer(Class module) { - this.module = module; - } - - public ProcessorConfigurer as(String... names) { - this.names = names; - return this; - } - - public ProcessorConfigurer profiles(String... profiles) { - this.profiles = profiles; - return this; - } - - public SinkConfigurer to(Class sink) { - build(); - return new SinkConfigurer(sink); - } - - public ProcessorConfigurer via(Class processor) { - build(); - return new ProcessorConfigurer(processor); - } - - private void build() { - childContext(this.module).config(this.names).profiles(this.profiles) - .input(incrementChannelName()).output(channelName()).build(); - } - - } - - private ChildContextBuilder childContext(Class type) { - return new ChildContextBuilder(new SpringApplicationBuilder(type, - SeedConfiguration.class).parent(AggregateBuilder.this.parent) - .bannerMode(Mode.OFF).web(false) - .initializers(new BeanPostProcessorInitializer())); - } - - private class ChildContextBuilder { - - private SpringApplicationBuilder builder; - - private String configName; - - private String input; - - private String output; - - 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 input(String input) { - this.input = input; - return this; - } - - public ChildContextBuilder output(String output) { - this.output = output; - return this; - } - - public void build() { - List args = new ArrayList(); - if (this.configName != null) { - args.add("--spring.config.name=" + this.configName); - } - if (this.input != null) { - args.add("--spring.cloud.stream.bindings.input=" + this.input); - } - if (this.output != null) { - args.add("--spring.cloud.stream.bindings.output=" + this.output); - } - this.builder.run(args.toArray(new String[0])); - } - - } - - private class BeanPostProcessorInitializer implements - ApplicationContextInitializer { - - @Override - public void initialize(ConfigurableApplicationContext applicationContext) { - } - - } - - @Configuration - @EnableAutoConfiguration - protected static class SeedConfiguration { - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateConfigurer.java deleted file mode 100644 index 80bad24c5..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateConfigurer.java +++ /dev/null @@ -1,27 +0,0 @@ -/* - * Copyright 2015 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; - -/** - * @author Dave Syer - * - */ -public interface AggregateConfigurer { - - void configure(AggregateBuilder builder); - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java index 847161835..4923b732e 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * 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. @@ -23,7 +23,6 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import org.springframework.cloud.stream.config.AggregateBuilderConfiguration; import org.springframework.cloud.stream.config.BinderFactoryConfiguration; import org.springframework.cloud.stream.config.BindingBeansRegistrar; import org.springframework.cloud.stream.config.ChannelBindingServiceConfiguration; @@ -44,8 +43,7 @@ import org.springframework.integration.config.EnableIntegration; @Documented @Inherited @Configuration -@Import({ChannelBindingServiceConfiguration.class, AggregateBuilderConfiguration.class, BindingBeansRegistrar.class, - BinderFactoryConfiguration.class}) +@Import({ChannelBindingServiceConfiguration.class, BindingBeansRegistrar.class, BinderFactoryConfiguration.class}) @EnableIntegration public @interface EnableBinding { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/AggregateBuilderConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/AggregateBuilderConfiguration.java deleted file mode 100644 index e59602ea3..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/AggregateBuilderConfiguration.java +++ /dev/null @@ -1,49 +0,0 @@ -/* - * Copyright 2015 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.config; - -import org.springframework.beans.factory.ListableBeanFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.CommandLineRunner; -import org.springframework.cloud.stream.aggregate.AggregateBuilder; -import org.springframework.cloud.stream.aggregate.AggregateConfigurer; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; - -/** - * @author Dave Syer - * - */ -@Configuration -public class AggregateBuilderConfiguration implements CommandLineRunner { - - @Autowired - private ListableBeanFactory beanFactory; - - @Bean - public AggregateBuilder aggregateBuilder() { - return new AggregateBuilder(); - } - - @Override - public void run(String... args) throws Exception { - for (AggregateConfigurer configurer : this.beanFactory.getBeansOfType(AggregateConfigurer.class).values()) { - configurer.configure(aggregateBuilder()); - } - aggregateBuilder().build(); - } -}