Add aggregate builder and sample applications

No need to specify channel names (just spring.cloud.streams.name). Example
app:

@SpringBootApplication
@EnableChannelBinding
public class ExtendedApplication implements AggregateConfigurer {

	@Override
	public void configure(AggregateBuilder builder) {
		builder
		.from(TimeSource.class).as("source")
		.via(LoggingTransformer.class)
		.via(LoggingTransformer.class).profiles("other")
		.to(LogSink.class);
	}
}

Fixes gh-2

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/EnableChannelBinding.java

@@ -23,8 +23,9 @@ import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
-import org.springframework.cloud.streams.config.LifecycleConfiguration;
+import org.springframework.cloud.streams.config.AggregateBuilderConfiguration;
 import org.springframework.cloud.streams.config.ChannelBindingAdapterConfiguration;
+import org.springframework.cloud.streams.config.LifecycleConfiguration;
 import org.springframework.cloud.streams.config.RabbitServiceConfiguration;
 import org.springframework.cloud.streams.config.RedisServiceConfiguration;
 import org.springframework.context.annotation.Configuration;
@@ -40,7 +41,8 @@ import org.springframework.context.annotation.Import;
 @Inherited
 @Configuration
 @Import({ RedisServiceConfiguration.class, RabbitServiceConfiguration.class,
-		ChannelBindingAdapterConfiguration.class, LifecycleConfiguration.class })
+	ChannelBindingAdapterConfiguration.class, LifecycleConfiguration.class,
+	AggregateBuilderConfiguration.class })
 public @interface EnableChannelBinding {
 
 }

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/aggregate/AggregateBuilder.java

@@ -0,0 +1,257 @@
+/*
+ * 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.streams.aggregate;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+
+import org.springframework.beans.BeansException;
+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 {
+
+	public static final String DEFAULT_NAME = "application";
+
+	private ConfigurableApplicationContext parent;
+
+	private int index = 0;
+
+	private String streamName = "stream";
+
+	private HashSet<SinkConfigurer> sinks = new HashSet<SinkConfigurer>();
+
+	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)
+				.showBanner(false).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<String> args = new ArrayList<String>();
+			if (this.configName != null) {
+				args.add("--spring.config.name=" + this.configName);
+			}
+			if (this.input != null) {
+				args.add("--spring.cloud.channels.inputChannelName=" + this.input);
+			}
+			if (this.output != null) {
+				args.add("--spring.cloud.channels.outputChannelName=" + this.output);
+			}
+			this.builder.run(args.toArray(new String[0]));
+		}
+
+	}
+
+	private class BeanPostProcessorInitializer implements
+	ApplicationContextInitializer<ConfigurableApplicationContext> {
+
+		@Override
+		public void initialize(ConfigurableApplicationContext applicationContext) {
+		}
+
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	protected static class SeedConfiguration {
+	}
+
+}

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/aggregate/AggregateConfigurer.java

@@ -0,0 +1,27 @@
+/*
+ * 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.streams.aggregate;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public interface AggregateConfigurer {
+
+	void configure(AggregateBuilder builder);
+
+}

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/AggregateBuilderConfiguration.java

@@ -0,0 +1,50 @@
+/*
+ * 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.streams.config;
+
+import org.springframework.beans.factory.ListableBeanFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.cloud.streams.aggregate.AggregateBuilder;
+import org.springframework.cloud.streams.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();
+	}
+
+}

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfiguration.java

@@ -32,6 +32,7 @@ import org.springframework.beans.factory.BeanFactoryUtils;
 import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.SearchStrategy;
 import org.springframework.cloud.streams.adapter.ChannelBindingAdapter;
 import org.springframework.cloud.streams.adapter.ChannelLocator;
 import org.springframework.cloud.streams.adapter.Input;
@@ -104,8 +105,7 @@ public class ChannelBindingAdapterConfiguration {
 
 	protected Collection<OutputChannelBinding> getOutputChannels() {
 		Set<OutputChannelBinding> channels = new LinkedHashSet<OutputChannelBinding>();
-		String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(
-				this.beanFactory, MessageChannel.class);
+		String[] names = this.beanFactory.getBeanNamesForType(MessageChannel.class);
 		for (String name : names) {
 			if (name.startsWith("output")) {
 				channels.add(new OutputChannelBinding(name));
@@ -116,8 +116,7 @@ public class ChannelBindingAdapterConfiguration {
 
 	protected Collection<InputChannelBinding> getInputChannels() {
 		Set<InputChannelBinding> channels = new LinkedHashSet<InputChannelBinding>();
-		String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(
-				this.beanFactory, MessageChannel.class);
+		String[] names = this.beanFactory.getBeanNamesForType(MessageChannel.class);
 		for (String name : names) {
 			if (name.startsWith("input")) {
 				channels.add(new InputChannelBinding(name));
@@ -174,7 +173,7 @@ public class ChannelBindingAdapterConfiguration {
 	}
 
 	@Configuration
-	@ConditionalOnMissingBean(ChannelBindingProperties.class)
+	@ConditionalOnMissingBean(value=ChannelBindingProperties.class, search=SearchStrategy.CURRENT)
 	protected static class ModulePropertiesConfiguration {
 		@Bean(name = "spring.cloud.channels.CONFIGURATION_PROPERTIES")
 		public ChannelBindingProperties moduleProperties() {

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/RabbitServiceConfiguration.java

@@ -18,6 +18,7 @@ package org.springframework.cloud.streams.config;
 
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.cloud.Cloud;
 import org.springframework.cloud.CloudFactory;
 import org.springframework.context.annotation.Bean;
@@ -35,6 +36,7 @@ import org.springframework.xd.dirt.integration.rabbit.RabbitMessageBus;
  */
 @Configuration
 @ConditionalOnClass(RabbitMessageBus.class)
+@ConditionalOnMissingBean(RabbitMessageBus.class)
 @ImportResource({ "classpath*:/META-INF/spring-xd/bus/rabbit-bus.xml",
 "classpath*:/META-INF/spring-xd/analytics/rabbit-analytics.xml" })
 @PropertySource("classpath:/META-INF/spring-cloud-streams/rabbit-bus.properties")

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/RedisServiceConfiguration.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.streams.config;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.cloud.Cloud;
 import org.springframework.cloud.CloudFactory;
 import org.springframework.context.annotation.Bean;
@@ -35,6 +36,7 @@ import org.springframework.xd.dirt.integration.redis.RedisMessageBus;
  */
 @Configuration
 @ConditionalOnClass(RedisMessageBus.class)
+@ConditionalOnMissingBean(RedisMessageBus.class)
 @ImportResource({ "classpath*:/META-INF/spring-xd/bus/redis-bus.xml",
 "classpath*:/META-INF/spring-xd/analytics/redis-analytics.xml" })
 @PropertySource("classpath:/META-INF/spring-cloud-streams/redis-bus.properties")