Add support for @EnableModule, @Input and @Output

* enable the component model based on @EnableModule with existing functionality
* any field annotated with Input/Output will trigger the creation and injection of a channel
* update examples and tests
* update README

README.adoc

@@ -2,15 +2,14 @@
 
 image::https://travis-ci.org/spring-cloud/spring-cloud-streams.svg?branch=master[Build Status, link=https://travis-ci.org/spring-cloud/spring-cloud-streams]
 
-This project allows a user to develop and run messaging microservices using Spring Integration and run them locally, or in the cloud, or even on Spring XD. Just create `MessageChannels` "input" and/or "output" and add `@EnableChannelBinding` and run your app as a Spring Boot app (single application context).  You just need to connect to the physical broker for the bus, which is automatic if the relevant bus implementation is available on the classpath. The sample uses Redis.
+This project allows a user to develop and run messaging microservices using Spring Integration and run them locally, or in the cloud, or even on Spring XD. Just create `MessageChannels` annotated with `@Input` and `@Output` and add `@EnableModule` and run your app as a Spring Boot app (single application context).  You just need to connect to the physical broker for the bus, which is automatic if the relevant bus implementation is available on the classpath. The sample uses Redis.
 
 Here's a sample source module (output channel only):
 
 [source,java]
 ----
 @SpringBootApplication
-@EnableChannelBinding
-@ComponentScan(basePackageClasses=ModuleDefinition.class)
+@ComponentScan(basePackageClasses=TimeSource.class)
 public class ModuleApplication {
 
   public static void main(String[] args) throws InterruptedException {
@@ -20,15 +19,14 @@ public class ModuleApplication {
 }
 
 @Configuration
-public class ModuleDefinition {
+@EnableModule
+public class TimeSource {
 
   @Value("${format}")
   private String format;
 
-  @Bean
-  public MessageChannel output() {
-    return new DirectChannel();
-  }
+  @Input
+  public MessageChannel output;
 
   @Bean
   @InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
@@ -54,7 +52,8 @@ To be deployable as an XD module in a "traditional" way you need `/config/*.prop
 
 == Multiple Input or Output Channels
 
-A module can have multiple input or output channels. Instead of just one channel named "input" or "output" you can add multiple `MessageChannel` beans named `input.*` or `output.*` and the names are converted to external channel names on the broker. The external channel names are the `spring.cloud.streams.[input|output]ChannelName` plus the `MessageChannel` bean name, period separated. In addition, the bean name can be `input.[queue|topic|tap]:*` or `output.[queue|topic]:*` (i.e. with a channel type as a colon-separated prefix), and the semantics of the external bus channel changes accordingly (a tap is like a topic). For example, you can have two `MessageChannels` called "output" and "output.topic:foo" in a module with `outputChannelName=bar`, and the result is 2 external channels called "bar" and "topic:foo.bar".
+A module can have multiple input or output channels. Instead of just one channel named "input" or "output" you can add multiple `MessageChannel` beans
+ annotated with `@Input` and `@Output` and the names are converted to external channel names on the broker. The external channel names are the `spring.cloud.streams.[input|output]ChannelName` plus the `MessageChannel` bean name, period separated. In addition, the bean name can be `input.[queue|topic|tap]:*` or `output.[queue|topic]:*` (i.e. with a channel type as a colon-separated prefix), and the semantics of the external bus channel changes accordingly (a tap is like a topic). For example, you can have two `MessageChannels` called "output" and "output.topic:foo" in a module with `outputChannelName=bar`, and the result is 2 external channels called "bar" and "topic:foo.bar".
 
 == Samples
 

spring-cloud-streams-samples/double/src/main/java/config/SinkModuleDefinition.java

@@ -18,29 +18,23 @@ package config;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.cloud.streams.EnableChannelBinding;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.annotation.MessageEndpoint;
+
+import org.springframework.cloud.streams.annotation.EnableModule;
+import org.springframework.cloud.streams.annotation.Input;
 import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.messaging.MessageChannel;
 
 /**
  * @author Dave Syer
- *
+ * @author Marius Bogoevici
  */
-@Configuration
-@EnableChannelBinding
-@MessageEndpoint
+@EnableModule
 public class SinkModuleDefinition {
 
 	private static Logger logger = LoggerFactory.getLogger(SinkModuleDefinition.class);
 
-	@Bean
-	public MessageChannel input() {
-		return new DirectChannel();
-	}
+	@Input
+	public MessageChannel input;
 
 	@ServiceActivator(inputChannel="input")
 	public void loggerSink(Object payload) {

spring-cloud-streams-samples/double/src/main/java/config/SourceModuleDefinition.java

@@ -20,31 +20,27 @@ import java.text.SimpleDateFormat;
 import java.util.Date;
 
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.cloud.streams.EnableChannelBinding;
+import org.springframework.cloud.streams.annotation.EnableModule;
+import org.springframework.cloud.streams.annotation.Output;
 import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.InboundChannelAdapter;
 import org.springframework.integration.annotation.Poller;
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.support.GenericMessage;
 
 /**
  * @author Dave Syer
- *
+ * @author Marius Bogoevici
  */
-@Configuration
-@EnableChannelBinding
+@EnableModule
 public class SourceModuleDefinition {
 
 	@Value("${format:YYYY/MM/dd hh:mm:ss}")
 	private String format;
 
-	@Bean
-	public MessageChannel output() {
-		return new DirectChannel();
-	}
+	@Output
+	public MessageChannel output;
 
 	@Bean
 	@InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))

spring-cloud-streams-samples/double/src/main/java/demo/DoubleApplication.java

@@ -2,7 +2,6 @@ package demo;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.cloud.streams.EnableChannelBinding;
 import org.springframework.cloud.streams.aggregate.AggregateBuilder;
 import org.springframework.cloud.streams.aggregate.AggregateConfigurer;
 
@@ -10,7 +9,6 @@ import config.SinkModuleDefinition;
 import config.SourceModuleDefinition;
 
 @SpringBootApplication
-@EnableChannelBinding
 public class DoubleApplication implements AggregateConfigurer {
 
 	@Override

spring-cloud-streams-samples/extended/src/main/java/extended/ExtendedApplication.java

@@ -2,7 +2,6 @@ package extended;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.cloud.streams.EnableChannelBinding;
 import org.springframework.cloud.streams.aggregate.AggregateBuilder;
 import org.springframework.cloud.streams.aggregate.AggregateConfigurer;
 
@@ -11,7 +10,6 @@ import source.TimeSource;
 import transform.LoggingTransformer;
 
 @SpringBootApplication
-@EnableChannelBinding
 public class ExtendedApplication implements AggregateConfigurer {
 
 	@Override

spring-cloud-streams-samples/sink/src/main/java/sink/LogSink.java

@@ -18,29 +18,23 @@ package sink;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.cloud.streams.EnableChannelBinding;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.annotation.MessageEndpoint;
+
+import org.springframework.cloud.streams.annotation.EnableModule;
+import org.springframework.cloud.streams.annotation.Input;
 import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.messaging.MessageChannel;
 
 /**
  * @author Dave Syer
  *
  */
-@Configuration
-@EnableChannelBinding
-@MessageEndpoint
+@EnableModule
 public class LogSink {
 
 	private static Logger logger = LoggerFactory.getLogger(LogSink.class);
 
-	@Bean
-	public MessageChannel input() {
-		return new DirectChannel();
-	}
+	@Input
+	public MessageChannel input;
 
 	@ServiceActivator(inputChannel="input")
 	public void loggerSink(Object payload) {

spring-cloud-streams-samples/source/src/main/java/source/TimeSource.java

@@ -21,7 +21,8 @@ import java.util.Date;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.cloud.streams.EnableChannelBinding;
+import org.springframework.cloud.streams.annotation.EnableModule;
+import org.springframework.cloud.streams.annotation.Output;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.InboundChannelAdapter;
@@ -35,18 +36,15 @@ import org.springframework.messaging.support.GenericMessage;
  * @author Dave Syer
  *
  */
-@Configuration
-@EnableChannelBinding
+@EnableModule
 @EnableConfigurationProperties(TimeSourceOptionsMetadata.class)
 public class TimeSource {
 
 	@Autowired
 	private TimeSourceOptionsMetadata options;
 
-	@Bean
-	public MessageChannel output() {
-		return new DirectChannel();
-	}
+	@Output
+	public MessageChannel output;
 
 	@Bean
 	@InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))

spring-cloud-streams-samples/source/src/main/java/source/TimeSourceOptionsMetadata.java

@@ -19,9 +19,6 @@ package source;
 import javax.validation.constraints.Min;
 import javax.validation.constraints.Pattern;
 
-import org.springframework.xd.module.options.mixins.MaxMessagesDefaultOneMixin;
-import org.springframework.xd.module.options.mixins.PeriodicTriggerMixin;
-import org.springframework.xd.module.options.spi.Mixin;
 import org.springframework.xd.module.options.validation.DateFormat;
 
 /**
@@ -30,7 +27,6 @@ import org.springframework.xd.module.options.validation.DateFormat;
  * @author Eric Bottard
  * @author Gary Russell
  */
-@Mixin({ PeriodicTriggerMixin.class, MaxMessagesDefaultOneMixin.class })
 public class TimeSourceOptionsMetadata {
 
 	/**

spring-cloud-streams-samples/transform/src/main/java/transform/LoggingTransformer.java

@@ -18,11 +18,10 @@ package transform;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
 import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.cloud.streams.EnableChannelBinding;
+import org.springframework.cloud.streams.annotation.EnableModule;
 import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.annotation.MessageEndpoint;
 import org.springframework.integration.annotation.ServiceActivator;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.messaging.MessageChannel;
@@ -32,9 +31,7 @@ import org.springframework.messaging.SubscribableChannel;
  * @author Dave Syer
  *
  */
-@Configuration
-@EnableChannelBinding
-@MessageEndpoint
+@EnableModule
 @ConfigurationProperties("module.logging")
 public class LoggingTransformer {
 

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/ChannelBindingAdapter.java

@@ -53,8 +53,11 @@ import org.springframework.xd.dirt.integration.bus.MessageBus;
 import org.springframework.xd.dirt.integration.bus.XdHeaders;
 
 /**
+ * Binds input/output channels to the bus.
+ *
  * @author Mark Fisher
  * @author Dave Syer
+ * @author Marius Bogoevici
  */
 @ManagedResource
 public class ChannelBindingAdapter implements Lifecycle, ApplicationContextAware {
@@ -77,9 +80,7 @@ public class ChannelBindingAdapter implements Lifecycle, ApplicationContextAware
 
 	private ConfigurableApplicationContext applicationContext;
 
-	private ChannelLocator inputChannelLocator;
-
-	private ChannelLocator outputChannelLocator;
+	private ChannelLocator channelLocator;
 
 	private DestinationResolver<MessageChannel> channelResolver;
 
@@ -88,16 +89,11 @@ public class ChannelBindingAdapter implements Lifecycle, ApplicationContextAware
 	public ChannelBindingAdapter(ChannelBindingProperties module, MessageBus messageBus) {
 		this.module = module;
 		this.messageBus = messageBus;
-		this.inputChannelLocator = new DefaultChannelLocator(module);
-		this.outputChannelLocator = new DefaultChannelLocator(module);
+		this.channelLocator = new DefaultChannelLocator(module);
 	}
 
-	public void setInputChannelLocator(ChannelLocator channelLocator) {
-		this.inputChannelLocator = channelLocator;
-	}
-
-	public void setOutputChannelLocator(ChannelLocator channelLocator) {
-		this.outputChannelLocator = channelLocator;
+	public void setChannelLocator(ChannelLocator channelLocator) {
+		this.channelLocator = channelLocator;
 	}
 
 	@Override
@@ -271,7 +267,7 @@ public class ChannelBindingAdapter implements Lifecycle, ApplicationContextAware
 			MessageChannel outputChannel = this.channelResolver.resolveDestination(binding.getLocalName());
 			bindMessageProducer(outputChannel, name, this.module.getProducerProperties());
 			if (binding.isTapped()) {
-				String tapChannelName = this.outputChannelLocator.tap(name);
+				String tapChannelName = this.channelLocator.tap(name);
 				binding.setTapChannelName(tapChannelName);
 				// tappableChannels.put(tapChannelName, outputChannel);
 				// if (isTapActive(tapChannelName)) {
@@ -299,7 +295,7 @@ public class ChannelBindingAdapter implements Lifecycle, ApplicationContextAware
 		logger.info("Locating channels");
 		boolean located = true;
 		for (OutputChannelBinding binding : this.outputChannels) {
-			String name = this.outputChannelLocator.locate(binding.getLocalName());
+			String name = this.channelLocator.locate(binding.getLocalName());
 			if (name == null) {
 				logger.info("No channel found for: " + binding.getLocalName());
 				located = false;
@@ -308,7 +304,7 @@ public class ChannelBindingAdapter implements Lifecycle, ApplicationContextAware
 			this.bindings.put(binding.getRemoteName(), name);
 		}
 		for (InputChannelBinding binding : this.inputChannels) {
-			String name = this.inputChannelLocator.locate(binding.getLocalName());
+			String name = this.channelLocator.locate(binding.getLocalName());
 			if (name == null) {
 				logger.info("No channel found for: " + binding.getLocalName());
 				located = false;

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/annotation/EnableModule.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.streams;
+package org.springframework.cloud.streams.annotation;
 
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;
@@ -24,25 +24,30 @@ import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
 import org.springframework.cloud.streams.config.AggregateBuilderConfiguration;
-import org.springframework.cloud.streams.config.ChannelBindingAdapterConfiguration;
+import org.springframework.cloud.streams.config.EnableModuleConfiguration;
 import org.springframework.cloud.streams.config.LifecycleConfiguration;
+import org.springframework.cloud.streams.config.ChannelBindingAdapterConfiguration;
 import org.springframework.cloud.streams.config.RabbitServiceConfiguration;
 import org.springframework.cloud.streams.config.RedisServiceConfiguration;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
+import org.springframework.integration.annotation.MessageEndpoint;
 
 /**
- * @author Dave Syer
+ * Annotation that identifies a class as a module.
  *
+ * @author Dave Syer
+ * @author Marius Bogoevici
  */
 @Target(ElementType.TYPE)
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 @Inherited
 @Configuration
+@MessageEndpoint
 @Import({ RedisServiceConfiguration.class, RabbitServiceConfiguration.class,
 	ChannelBindingAdapterConfiguration.class, LifecycleConfiguration.class,
-	AggregateBuilderConfiguration.class })
-public @interface EnableChannelBinding {
+	AggregateBuilderConfiguration.class, EnableModuleConfiguration.class})
+public @interface EnableModule {
 
 }

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/annotation/Input.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.streams.adapter;
+package org.springframework.cloud.streams.annotation;
 
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;
@@ -23,15 +23,17 @@ import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 
 /**
- * Qualifier annotation for a bean relating input channels.
+ * Indicates that an input channel will be created and injected by the framework.
  *
  * @author Dave Syer
+ * @author Marius Bogoevici
  */
-@Qualifier
-@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER, ElementType.TYPE,
+
+@Target({ ElementType.FIELD, ElementType.METHOD,
 	ElementType.ANNOTATION_TYPE })
 @Retention(RetentionPolicy.RUNTIME)
 @Inherited

spring-cloud-streams/src/main/java/org/springframework/cloud/streams/annotation/Output.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.streams.adapter;
+package org.springframework.cloud.streams.annotation;
 
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;
@@ -23,16 +23,17 @@ import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 
 /**
- * Qualifier annotation for a bean relating output channels.
+ * Indicates that an output channel will be created and injected by the framework.
  *
  * @author Dave Syer
+ * @author Marius Bogoevici
  */
-@Qualifier
-@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.PARAMETER, ElementType.TYPE,
-	ElementType.ANNOTATION_TYPE })
+
+@Target({ ElementType.FIELD, ElementType.METHOD})
 @Retention(RetentionPolicy.RUNTIME)
 @Inherited
 @Documented

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

@@ -25,6 +25,7 @@ import java.util.Set;
 
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
+
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.target.LazyInitTargetSource;
 import org.springframework.beans.factory.BeanFactoryUtils;
@@ -33,9 +34,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.cloud.streams.adapter.ChannelBindingAdapter;
 import org.springframework.cloud.streams.adapter.ChannelLocator;
-import org.springframework.cloud.streams.adapter.Input;
 import org.springframework.cloud.streams.adapter.InputChannelBinding;
-import org.springframework.cloud.streams.adapter.Output;
 import org.springframework.cloud.streams.adapter.OutputChannelBinding;
 import org.springframework.cloud.streams.endpoint.ChannelsEndpoint;
 import org.springframework.context.ApplicationContext;
@@ -54,6 +53,7 @@ import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec;
 /**
  * @author Dave Syer
  * @author David Turanski
+ * @author Marius Bogoevici
  */
 @Configuration
 public class ChannelBindingAdapterConfiguration {
@@ -64,13 +64,7 @@ public class ChannelBindingAdapterConfiguration {
 	@Autowired
 	private ListableBeanFactory beanFactory;
 
-	@Autowired(required = false)
-	@Input
-	private ChannelLocator inputChannelLocator;
-
-	@Autowired(required = false)
-	@Output
-	private ChannelLocator outputChannelLocator;
+	private ChannelLocator channelLocator;
 
 	@Autowired
 	private MessageBus messageBus;
@@ -80,11 +74,8 @@ public class ChannelBindingAdapterConfiguration {
 		ChannelBindingAdapter adapter = new ChannelBindingAdapter(this.module, this.messageBus);
 		adapter.setOutputChannels(getOutputChannels());
 		adapter.setInputChannels(getInputChannels());
-		if (this.inputChannelLocator != null) {
-			adapter.setInputChannelLocator(this.inputChannelLocator);
-		}
-		if (this.outputChannelLocator != null) {
-			adapter.setOutputChannelLocator(this.outputChannelLocator);
+		if (this.channelLocator!=null) {
+			adapter.setChannelLocator(this.channelLocator);
 		}
 		return adapter;
 	}

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

@@ -0,0 +1,51 @@
+/*
+ * 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.FactoryBean;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.messaging.MessageChannel;
+
+/**
+ * {@link FactoryBean} for creating channels for fields annotated with
+ * {@link org.springframework.cloud.streams.annotation.Input} and
+ * {@link org.springframework.cloud.streams.annotation.Output}.
+ *
+ * @author Marius Bogoevici
+ */
+public class DirectChannelFactoryBean implements FactoryBean<MessageChannel> {
+
+	private DirectChannel directChannel;
+
+	@Override
+	public synchronized MessageChannel getObject() throws Exception {
+		if (directChannel == null) {
+			directChannel = new DirectChannel();
+		}
+		return directChannel;
+	}
+
+	@Override
+	public Class<?> getObjectType() {
+		return MessageChannel.class;
+	}
+
+	@Override
+	public boolean isSingleton() {
+		return true;
+	}
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author Marius Bogoevici
+ */
+@Configuration
+public class EnableModuleConfiguration {
+
+	@Bean
+	public ModulePostProcessor modulePostProcessor() {
+		return new ModulePostProcessor();
+	}
+
+}

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

@@ -0,0 +1,105 @@
+/*
+ * 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 java.lang.reflect.Field;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.beans.factory.support.AutowireCandidateQualifier;
+import org.springframework.beans.factory.support.BeanDefinitionRegistry;
+import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor;
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.cloud.streams.annotation.EnableModule;
+import org.springframework.cloud.streams.annotation.Input;
+import org.springframework.cloud.streams.annotation.Output;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ReflectionUtils;
+
+/**
+ * @author Marius Bogoevici
+ */
+public class ModulePostProcessor implements BeanDefinitionRegistryPostProcessor,BeanPostProcessor, ApplicationContextAware {
+
+	private ApplicationContext applicationContext;
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		this.applicationContext = applicationContext;
+	}
+
+	@Override
+	public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
+		String[] beanDefinitionNames = registry.getBeanDefinitionNames();
+		for (String beanDefinitionName : beanDefinitionNames) {
+			BeanDefinition beanDefinition = registry.getBeanDefinition(beanDefinitionName);
+			String beanClassName = beanDefinition.getBeanClassName();
+			try {
+				if (beanClassName != null && ClassUtils.forName(beanClassName,null).isAnnotationPresent((EnableModule.class))) {
+					ReflectionUtils.doWithFields(ClassUtils.forName(beanClassName, null), new ReflectionUtils.FieldCallback() {
+						@Override
+						public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException {
+							if (field.isAnnotationPresent(Input.class)) {
+								RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(DirectChannelFactoryBean.class);
+								rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(Input.class));
+								registry.registerBeanDefinition(field.getName(), rootBeanDefinition);
+							}
+							if (field.isAnnotationPresent(Output.class)) {
+								RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(DirectChannelFactoryBean.class);
+								rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(Output.class));
+								registry.registerBeanDefinition(field.getName(), rootBeanDefinition);
+							}
+						}
+					});
+				}
+			} catch (ClassNotFoundException e) {
+				throw new IllegalStateException(e);
+			}
+		}
+	}
+
+	@Override
+	public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
+	}
+
+	@Override
+	public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+		if (AnnotationUtils.findAnnotation(bean.getClass(), EnableModule.class) != null) {
+			ReflectionUtils.doWithFields(bean.getClass(), new ReflectionUtils.FieldCallback() {
+				@Override
+				public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException {
+					if (field.isAnnotationPresent(Output.class) || field.isAnnotationPresent(Input.class)) {
+						// TODO: Do not rely on field name for injection
+						ReflectionUtils.makeAccessible(field);
+						ReflectionUtils.setField(field, bean, applicationContext.getBean(field.getName()));
+					}
+				}
+			});
+		}
+		return bean;
+	}
+
+	@Override
+	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+		return bean;
+	}
+}

spring-xd-runner/src/test/java/org/springframework/cloud/streams/xd/ChannelBindingAdapterConfigurationTests.java

@@ -26,6 +26,7 @@ import java.util.List;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.support.DefaultListableBeanFactory;
 import org.springframework.boot.test.SpringApplicationConfiguration;

spring-xd-samples/sink/src/main/java/config/LoggerSink.java

@@ -18,27 +18,26 @@ package config;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.context.annotation.Bean;
+
+import org.springframework.cloud.streams.annotation.EnableModule;
+import org.springframework.cloud.streams.annotation.Input;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.MessageEndpoint;
 import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.messaging.MessageChannel;
 
 /**
  * @author Dave Syer
+ * @author Marius Bogoevici
  *
  */
-@Configuration
-@MessageEndpoint
-public class ModuleDefinition {
+@EnableModule
+public class LoggerSink {
 	
-	private static Logger logger = LoggerFactory.getLogger(ModuleDefinition.class);
+	private static Logger logger = LoggerFactory.getLogger(LoggerSink.class);
 
-	@Bean
-	public MessageChannel input() {
-		return new DirectChannel();
-	}
+	@Input
+	private MessageChannel input;
 
 	@ServiceActivator(inputChannel="input")
 	public void loggerSink(Object payload) {

spring-xd-samples/sink/src/main/java/demo/SinkApplication.java

@@ -2,14 +2,13 @@ package demo;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.cloud.streams.EnableChannelBinding;
+import org.springframework.cloud.streams.annotation.EnableModule;
 import org.springframework.context.annotation.ComponentScan;
 
-import config.ModuleDefinition;
+import config.LoggerSink;
 
 @SpringBootApplication
-@EnableChannelBinding
-@ComponentScan(basePackageClasses=ModuleDefinition.class)
+@ComponentScan(basePackageClasses = LoggerSink.class)
 public class SinkApplication {
 
 	public static void main(String[] args) throws InterruptedException {

spring-xd-samples/source-xml/src/main/java/demo/ModuleApplication.java

@@ -2,12 +2,12 @@ package demo;
 
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.builder.SpringApplicationBuilder;
-import org.springframework.cloud.streams.EnableChannelBinding;
+import org.springframework.cloud.streams.annotation.EnableModule;
 import org.springframework.context.annotation.ImportResource;
 import org.springframework.context.annotation.PropertySource;
 
 @SpringBootApplication
-@EnableChannelBinding
+@EnableModule
 @ImportResource("classpath:/config/ticker.xml")
 @PropertySource("classpath:/config/ticker.properties")
 public class ModuleApplication {

spring-xd-samples/source/src/main/java/config/ModuleDefinition.java

@@ -20,6 +20,8 @@ import java.text.SimpleDateFormat;
 import java.util.Date;
 
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.cloud.streams.annotation.EnableModule;
+import org.springframework.cloud.streams.annotation.Output;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.annotation.InboundChannelAdapter;
@@ -31,21 +33,20 @@ import org.springframework.messaging.support.GenericMessage;
 
 /**
  * @author Dave Syer
- *
+ * @author Marius Bogoevici
  */
-@Configuration
+@EnableModule
 public class ModuleDefinition {
 
 	@Value("${format}")
 	private String format;
 
-	@Bean
-	public MessageChannel output() {
-		return new DirectChannel();
-	}
+	@Output
+	public MessageChannel output;
 
 	@Bean
-	@InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
+	@InboundChannelAdapter(value = "output", autoStartup = "false",
+			poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
 	public MessageSource<String> timerMessageSource() {
 		return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
 	}

spring-xd-samples/source/src/main/java/demo/SourceApplication.java

@@ -2,13 +2,13 @@ package demo;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.cloud.streams.EnableChannelBinding;
+import org.springframework.cloud.streams.annotation.EnableModule;
 import org.springframework.context.annotation.ComponentScan;
 
 import config.ModuleDefinition;
 
 @SpringBootApplication
-@EnableChannelBinding
+@EnableModule
 @ComponentScan(basePackageClasses=ModuleDefinition.class)
 public class SourceApplication {
 

spring-xd-samples/tap/src/main/java/config/TappingLoggingSink.java

@@ -18,27 +18,23 @@ package config;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.annotation.MessageEndpoint;
+
+import org.springframework.cloud.streams.annotation.EnableModule;
+import org.springframework.cloud.streams.annotation.Input;
 import org.springframework.integration.annotation.ServiceActivator;
-import org.springframework.integration.channel.DirectChannel;
 import org.springframework.messaging.MessageChannel;
 
 /**
  * @author Dave Syer
- *
+ * @author Marius Bogoevici
  */
-@Configuration
-@MessageEndpoint
-public class ModuleDefinition {
+@EnableModule
+public class TappingLoggingSink {
 	
-	private static Logger logger = LoggerFactory.getLogger(ModuleDefinition.class);
+	private static Logger logger = LoggerFactory.getLogger(TappingLoggingSink.class);
 
-	@Bean
-	public MessageChannel input() {
-		return new DirectChannel();
-	}
+	@Input
+	public MessageChannel input;
 
 	@ServiceActivator(inputChannel="input")
 	public void loggerSink(Object payload) {

spring-xd-samples/tap/src/main/java/demo/TapApplication.java

@@ -2,14 +2,14 @@ package demo;
 
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.cloud.streams.EnableChannelBinding;
+import org.springframework.cloud.streams.annotation.EnableModule;
 import org.springframework.context.annotation.ComponentScan;
 
-import config.ModuleDefinition;
+import config.TappingLoggingSink;
 
 @SpringBootApplication
-@EnableChannelBinding
-@ComponentScan(basePackageClasses=ModuleDefinition.class)
+@EnableModule
+@ComponentScan(basePackageClasses= TappingLoggingSink.class)
 public class TapApplication {
 
 	public static void main(String[] args) throws InterruptedException {