From 23e4cbc7ac19a80c1214fe279599ed5d700cb1ca Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Thu, 9 Jul 2015 12:01:44 -0400 Subject: [PATCH] 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 | 17 ++- .../java/config/SinkModuleDefinition.java | 20 ++-- .../java/config/SourceModuleDefinition.java | 16 +-- .../src/main/java/demo/DoubleApplication.java | 2 - .../java/extended/ExtendedApplication.java | 2 - .../sink/src/main/java/sink/LogSink.java | 18 +-- .../src/main/java/source/TimeSource.java | 12 +- .../source/TimeSourceOptionsMetadata.java | 4 - .../java/transform/LoggingTransformer.java | 9 +- .../adapter/ChannelBindingAdapter.java | 24 ++-- .../EnableModule.java} | 15 ++- .../{adapter => annotation}/Input.java | 10 +- .../{adapter => annotation}/Output.java | 11 +- .../ChannelBindingAdapterConfiguration.java | 19 +--- .../config/DirectChannelFactoryBean.java | 51 +++++++++ .../config/EnableModuleConfiguration.java | 33 ++++++ .../streams/config/ModulePostProcessor.java | 105 ++++++++++++++++++ ...annelBindingAdapterConfigurationTests.java | 1 + ...{ModuleDefinition.java => LoggerSink.java} | 19 ++-- .../src/main/java/demo/SinkApplication.java | 7 +- .../src/main/java/demo/ModuleApplication.java | 4 +- .../main/java/config/ModuleDefinition.java | 15 +-- .../src/main/java/demo/SourceApplication.java | 4 +- ...efinition.java => TappingLoggingSink.java} | 22 ++-- .../src/main/java/demo/TapApplication.java | 8 +- 25 files changed, 299 insertions(+), 149 deletions(-) rename spring-cloud-streams/src/main/java/org/springframework/cloud/streams/{EnableChannelBinding.java => annotation/EnableModule.java} (80%) rename spring-cloud-streams/src/main/java/org/springframework/cloud/streams/{adapter => annotation}/Input.java (79%) rename spring-cloud-streams/src/main/java/org/springframework/cloud/streams/{adapter => annotation}/Output.java (78%) create mode 100644 spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/DirectChannelFactoryBean.java create mode 100644 spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/EnableModuleConfiguration.java create mode 100644 spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ModulePostProcessor.java rename spring-xd-samples/sink/src/main/java/config/{ModuleDefinition.java => LoggerSink.java} (76%) rename spring-xd-samples/tap/src/main/java/config/{ModuleDefinition.java => TappingLoggingSink.java} (67%) diff --git a/README.adoc b/README.adoc index 82a801b36..310b3c329 100644 --- a/README.adoc +++ b/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 diff --git a/spring-cloud-streams-samples/double/src/main/java/config/SinkModuleDefinition.java b/spring-cloud-streams-samples/double/src/main/java/config/SinkModuleDefinition.java index 81d8fc0e1..202432126 100644 --- a/spring-cloud-streams-samples/double/src/main/java/config/SinkModuleDefinition.java +++ b/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) { diff --git a/spring-cloud-streams-samples/double/src/main/java/config/SourceModuleDefinition.java b/spring-cloud-streams-samples/double/src/main/java/config/SourceModuleDefinition.java index c111f748f..90fd6cb1c 100644 --- a/spring-cloud-streams-samples/double/src/main/java/config/SourceModuleDefinition.java +++ b/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")) diff --git a/spring-cloud-streams-samples/double/src/main/java/demo/DoubleApplication.java b/spring-cloud-streams-samples/double/src/main/java/demo/DoubleApplication.java index 6cd23443c..6082ce8b6 100644 --- a/spring-cloud-streams-samples/double/src/main/java/demo/DoubleApplication.java +++ b/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 diff --git a/spring-cloud-streams-samples/extended/src/main/java/extended/ExtendedApplication.java b/spring-cloud-streams-samples/extended/src/main/java/extended/ExtendedApplication.java index 5065d2b64..8b8aca2f0 100644 --- a/spring-cloud-streams-samples/extended/src/main/java/extended/ExtendedApplication.java +++ b/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 diff --git a/spring-cloud-streams-samples/sink/src/main/java/sink/LogSink.java b/spring-cloud-streams-samples/sink/src/main/java/sink/LogSink.java index b28e4a03c..df329d83d 100644 --- a/spring-cloud-streams-samples/sink/src/main/java/sink/LogSink.java +++ b/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) { diff --git a/spring-cloud-streams-samples/source/src/main/java/source/TimeSource.java b/spring-cloud-streams-samples/source/src/main/java/source/TimeSource.java index 9826971e5..6d9953d49 100644 --- a/spring-cloud-streams-samples/source/src/main/java/source/TimeSource.java +++ b/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")) diff --git a/spring-cloud-streams-samples/source/src/main/java/source/TimeSourceOptionsMetadata.java b/spring-cloud-streams-samples/source/src/main/java/source/TimeSourceOptionsMetadata.java index 53481844a..5f184f6ae 100644 --- a/spring-cloud-streams-samples/source/src/main/java/source/TimeSourceOptionsMetadata.java +++ b/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 { /** diff --git a/spring-cloud-streams-samples/transform/src/main/java/transform/LoggingTransformer.java b/spring-cloud-streams-samples/transform/src/main/java/transform/LoggingTransformer.java index 539bec045..2d72eb18f 100644 --- a/spring-cloud-streams-samples/transform/src/main/java/transform/LoggingTransformer.java +++ b/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 { diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/ChannelBindingAdapter.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/ChannelBindingAdapter.java index 65e47633a..88f6d9183 100644 --- a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/ChannelBindingAdapter.java +++ b/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 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; diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/EnableChannelBinding.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/annotation/EnableModule.java similarity index 80% rename from spring-cloud-streams/src/main/java/org/springframework/cloud/streams/EnableChannelBinding.java rename to spring-cloud-streams/src/main/java/org/springframework/cloud/streams/annotation/EnableModule.java index 8fb515ac4..4e13c4adc 100644 --- a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/EnableChannelBinding.java +++ b/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 { } diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/Input.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/annotation/Input.java similarity index 79% rename from spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/Input.java rename to spring-cloud-streams/src/main/java/org/springframework/cloud/streams/annotation/Input.java index 22c4c43fb..85a2e2492 100644 --- a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/Input.java +++ b/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 diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/Output.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/annotation/Output.java similarity index 78% rename from spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/Output.java rename to spring-cloud-streams/src/main/java/org/springframework/cloud/streams/annotation/Output.java index 9d5b8b8a7..19297b76b 100644 --- a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/adapter/Output.java +++ b/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 diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfiguration.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfiguration.java index 621125900..ee39f67cb 100644 --- a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfiguration.java +++ b/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; } diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/DirectChannelFactoryBean.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/DirectChannelFactoryBean.java new file mode 100644 index 000000000..bc42f440d --- /dev/null +++ b/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 { + + 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; + } +} diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/EnableModuleConfiguration.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/EnableModuleConfiguration.java new file mode 100644 index 000000000..c6e3cbf12 --- /dev/null +++ b/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(); + } + +} diff --git a/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ModulePostProcessor.java b/spring-cloud-streams/src/main/java/org/springframework/cloud/streams/config/ModulePostProcessor.java new file mode 100644 index 000000000..19a519f0d --- /dev/null +++ b/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; + } +} diff --git a/spring-xd-runner/src/test/java/org/springframework/cloud/streams/xd/ChannelBindingAdapterConfigurationTests.java b/spring-xd-runner/src/test/java/org/springframework/cloud/streams/xd/ChannelBindingAdapterConfigurationTests.java index ad3b64237..f3de1e6e2 100644 --- a/spring-xd-runner/src/test/java/org/springframework/cloud/streams/xd/ChannelBindingAdapterConfigurationTests.java +++ b/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; diff --git a/spring-xd-samples/sink/src/main/java/config/ModuleDefinition.java b/spring-xd-samples/sink/src/main/java/config/LoggerSink.java similarity index 76% rename from spring-xd-samples/sink/src/main/java/config/ModuleDefinition.java rename to spring-xd-samples/sink/src/main/java/config/LoggerSink.java index 245cc76f5..c7b6311f7 100644 --- a/spring-xd-samples/sink/src/main/java/config/ModuleDefinition.java +++ b/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) { diff --git a/spring-xd-samples/sink/src/main/java/demo/SinkApplication.java b/spring-xd-samples/sink/src/main/java/demo/SinkApplication.java index af7b4e64f..c451d4661 100644 --- a/spring-xd-samples/sink/src/main/java/demo/SinkApplication.java +++ b/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 { diff --git a/spring-xd-samples/source-xml/src/main/java/demo/ModuleApplication.java b/spring-xd-samples/source-xml/src/main/java/demo/ModuleApplication.java index a646830eb..2a93f499d 100644 --- a/spring-xd-samples/source-xml/src/main/java/demo/ModuleApplication.java +++ b/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 { diff --git a/spring-xd-samples/source/src/main/java/config/ModuleDefinition.java b/spring-xd-samples/source/src/main/java/config/ModuleDefinition.java index 9d4e93c62..a505d26e4 100644 --- a/spring-xd-samples/source/src/main/java/config/ModuleDefinition.java +++ b/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 timerMessageSource() { return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date())); } diff --git a/spring-xd-samples/source/src/main/java/demo/SourceApplication.java b/spring-xd-samples/source/src/main/java/demo/SourceApplication.java index 55059897f..4b54973cd 100644 --- a/spring-xd-samples/source/src/main/java/demo/SourceApplication.java +++ b/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 { diff --git a/spring-xd-samples/tap/src/main/java/config/ModuleDefinition.java b/spring-xd-samples/tap/src/main/java/config/TappingLoggingSink.java similarity index 67% rename from spring-xd-samples/tap/src/main/java/config/ModuleDefinition.java rename to spring-xd-samples/tap/src/main/java/config/TappingLoggingSink.java index 245cc76f5..3d986e5b7 100644 --- a/spring-xd-samples/tap/src/main/java/config/ModuleDefinition.java +++ b/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) { diff --git a/spring-xd-samples/tap/src/main/java/demo/TapApplication.java b/spring-xd-samples/tap/src/main/java/demo/TapApplication.java index f2b5ba0d0..5c554ef57 100644 --- a/spring-xd-samples/tap/src/main/java/demo/TapApplication.java +++ b/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 {