From c09b292d7930a0ef5af5f2a50f0836b1576ee204 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Thu, 16 Jul 2015 15:10:41 +0100 Subject: [PATCH] Use @EnableModule(Channels.class) instead of field declarations Standard channel interface types of Sink, Source, Processor are provided, but user can create others with @Input and @Output MessageChannel methods. To get the declared channels in an app you can use the @ModuleChannels qualifier to inject the interface, e.g. @ModuleChannels(Cafe.class) injects the channels defined in the Cafe. --- docs/src/main/asciidoc/intro.adoc | 50 +++-- .../java/config/SinkModuleDefinition.java | 11 +- .../java/config/SourceModuleDefinition.java | 10 +- .../sink/src/main/java/sink/LogSink.java | 11 +- .../java/demo/ModuleApplicationTests.java | 25 ++- .../src/main/java/source/TimeSource.java | 10 +- .../main/java/config/TappingLoggingSink.java | 13 +- .../src/main/java/demo/TapApplication.java | 2 - .../java/transform/LoggingTransformer.java | 20 +- .../cloud/stream/annotation/EnableModule.java | 6 +- .../cloud/stream/annotation/Input.java | 6 +- .../stream/annotation/ModuleChannels.java | 44 +++++ .../cloud/stream/annotation/Output.java | 6 +- .../cloud/stream/annotation/Processor.java | 5 + .../cloud/stream/annotation/Sink.java | 12 ++ .../cloud/stream/annotation/Source.java | 15 ++ .../ChannelBindingAdapterConfiguration.java | 9 - .../config/EnableModuleConfiguration.java | 40 +++- .../stream/config/ModulePostProcessor.java | 100 ---------- .../utils/BeanDefinitionRegistryUtils.java | 49 ----- ...ageChannelBeanDefinitionRegistryUtils.java | 179 ++++++++++++++++++ ...annelBindingAdapterConfigurationTests.java | 15 +- 22 files changed, 390 insertions(+), 248 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/ModuleChannels.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Processor.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Sink.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Source.java delete mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ModulePostProcessor.java delete mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/BeanDefinitionRegistryUtils.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/MessageChannelBeanDefinitionRegistryUtils.java diff --git a/docs/src/main/asciidoc/intro.adoc b/docs/src/main/asciidoc/intro.adoc index db9ef9a6a..7cd3ee3cf 100644 --- a/docs/src/main/asciidoc/intro.adoc +++ b/docs/src/main/asciidoc/intro.adoc @@ -1,12 +1,11 @@ -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 `@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. +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 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 -@EnableModule -@ComponentScan(basePackageClasses=ModuleDefinition.class) +@ComponentScan(basePackageClasses=TimerSource.class) public class ModuleApplication { public static void main(String[] args) throws InterruptedException { @@ -16,18 +15,14 @@ public class ModuleApplication { } @Configuration -public class ModuleDefinition { +@EnableModule(Source.class) +public class TimerSource { @Value("${format}") private String format; @Bean - public MessageChannel output() { - return new DirectChannel(); - } - - @Bean - @InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1")) + @InboundChannelAdapter(value = Source.OUTPUT, autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1")) public MessageSource timerMessageSource() { return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date())); } @@ -46,9 +41,42 @@ spring: outputChannelName: ${spring.application.name:ticker} ---- +`@EnableModule` is parameterized by an interface (in this case `Source`) which declares input and output channels. `Source`, `Sink` and `Processor` are provided off the shelf, but you can define others. Here's the definition of `Source` + +[source,java] +---- +public interface Source { + @Output("output") + MessageChannel output(); +} +---- + +The `@Output` annotation is used to identify output channels (messages leaving the module) and `@Input` is used to identify input channels (messages entering the module). It is optionally parameterized by a channel name - if the name is not provided the method name is used instead. An implementation of the interface is created for you and can be used in the application context by autowiring it, e.g. into a test case: + +[source,java] +---- +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(classes = ModuleApplication.class) +@WebAppConfiguration +@DirtiesContext +public class ModuleApplicationTests { + + @Autowired + private Source source + + @Test + public void contextLoads() { + assertNotNull(this.sink.output()); + } + +} +---- + +NOTE: In this case there is only one `Source` in the application context so there is no need to qualify it when it is autowired. If there is ambiguity, e.g. if you are composing one module from some others, you can use `@ModuleChannels` qualifier to inject a specific channel set. The `@ModuleChannels` qualifier takes a parameter which is the class that carries the `@EnableModule` annotation (in this case the `TimerSource`). + == 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 all defined either as `@Input` and `@Output` methods in an interface (preferrable) or as bean definitions. Instead of just one channel named "input" or "output" you can add multiple `MessageChannel` methods annotated `input.*` or `output.*` and the names are converted to external channel names on the broker. The external channel names are the `spring.cloud.channels.[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-stream-samples/double/src/main/java/config/SinkModuleDefinition.java b/spring-cloud-stream-samples/double/src/main/java/config/SinkModuleDefinition.java index 188c22364..5276a61bf 100644 --- a/spring-cloud-stream-samples/double/src/main/java/config/SinkModuleDefinition.java +++ b/spring-cloud-stream-samples/double/src/main/java/config/SinkModuleDefinition.java @@ -18,25 +18,20 @@ package config; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import org.springframework.cloud.stream.annotation.EnableModule; -import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.Sink; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.messaging.MessageChannel; /** * @author Dave Syer * @author Marius Bogoevici */ -@EnableModule +@EnableModule(Sink.class) public class SinkModuleDefinition { private static Logger logger = LoggerFactory.getLogger(SinkModuleDefinition.class); - @Input - public MessageChannel input; - - @ServiceActivator(inputChannel="input") + @ServiceActivator(inputChannel=Sink.INPUT) public void loggerSink(Object payload) { logger.info("Received: " + payload); } diff --git a/spring-cloud-stream-samples/double/src/main/java/config/SourceModuleDefinition.java b/spring-cloud-stream-samples/double/src/main/java/config/SourceModuleDefinition.java index bb6a2ae09..a3c6c6d6b 100644 --- a/spring-cloud-stream-samples/double/src/main/java/config/SourceModuleDefinition.java +++ b/spring-cloud-stream-samples/double/src/main/java/config/SourceModuleDefinition.java @@ -21,29 +21,25 @@ import java.util.Date; import org.springframework.beans.factory.annotation.Value; import org.springframework.cloud.stream.annotation.EnableModule; -import org.springframework.cloud.stream.annotation.Output; +import org.springframework.cloud.stream.annotation.Source; import org.springframework.context.annotation.Bean; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.Poller; import org.springframework.integration.core.MessageSource; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; /** * @author Dave Syer * @author Marius Bogoevici */ -@EnableModule +@EnableModule(Source.class) public class SourceModuleDefinition { @Value("${format:YYYY/MM/dd hh:mm:ss}") private String format; - @Output - public MessageChannel output; - @Bean - @InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1")) + @InboundChannelAdapter(value = Source.OUTPUT, autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1")) public MessageSource timerMessageSource() { return () -> new GenericMessage<>(new SimpleDateFormat(this.format).format(new Date())); } diff --git a/spring-cloud-stream-samples/sink/src/main/java/sink/LogSink.java b/spring-cloud-stream-samples/sink/src/main/java/sink/LogSink.java index d7a1c92d1..1db4dad0a 100644 --- a/spring-cloud-stream-samples/sink/src/main/java/sink/LogSink.java +++ b/spring-cloud-stream-samples/sink/src/main/java/sink/LogSink.java @@ -18,25 +18,20 @@ package sink; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import org.springframework.cloud.stream.annotation.EnableModule; -import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.Sink; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.messaging.MessageChannel; /** * @author Dave Syer * */ -@EnableModule +@EnableModule(Sink.class) public class LogSink { private static Logger logger = LoggerFactory.getLogger(LogSink.class); - @Input - public MessageChannel input; - - @ServiceActivator(inputChannel="input") + @ServiceActivator(inputChannel=Sink.INPUT) public void loggerSink(Object payload) { logger.info("Received: " + payload); } diff --git a/spring-cloud-stream-samples/sink/src/test/java/demo/ModuleApplicationTests.java b/spring-cloud-stream-samples/sink/src/test/java/demo/ModuleApplicationTests.java index f2d65e89f..71263b9ec 100644 --- a/spring-cloud-stream-samples/sink/src/test/java/demo/ModuleApplicationTests.java +++ b/spring-cloud-stream-samples/sink/src/test/java/demo/ModuleApplicationTests.java @@ -1,11 +1,21 @@ package demo; +import static org.junit.Assert.assertNotNull; + import org.junit.Test; import org.junit.runner.RunWith; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.web.WebAppConfiguration; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.cloud.stream.annotation.ModuleChannels; +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.cloud.stream.annotation.Sink; +import org.springframework.cloud.stream.annotation.Source; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.web.WebAppConfiguration; + +import sink.LogSink; @RunWith(SpringJUnit4ClassRunner.class) @SpringApplicationConfiguration(classes = SinkApplication.class) @@ -13,8 +23,19 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @DirtiesContext public class ModuleApplicationTests { + @Autowired + @ModuleChannels(LogSink.class) + private Sink sink; + + @Autowired + private Sink same; + + @Output(Source.OUTPUT) + private MessageChannel output; + @Test public void contextLoads() { + assertNotNull(this.sink.input()); } } diff --git a/spring-cloud-stream-samples/source/src/main/java/source/TimeSource.java b/spring-cloud-stream-samples/source/src/main/java/source/TimeSource.java index 36894dcc1..c7148a76d 100644 --- a/spring-cloud-stream-samples/source/src/main/java/source/TimeSource.java +++ b/spring-cloud-stream-samples/source/src/main/java/source/TimeSource.java @@ -19,12 +19,11 @@ package source; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.annotation.EnableModule; -import org.springframework.cloud.stream.annotation.Output; +import org.springframework.cloud.stream.annotation.Source; import org.springframework.context.annotation.Bean; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.Poller; import org.springframework.integration.core.MessageSource; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; import java.text.SimpleDateFormat; @@ -35,18 +34,15 @@ import java.util.Date; * @author Glenn Renfro * */ -@EnableModule +@EnableModule(Source.class) @EnableConfigurationProperties(TimeSourceOptionsMetadata.class) public class TimeSource { @Autowired private TimeSourceOptionsMetadata options; - @Output - public MessageChannel output; - @Bean - @InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1")) + @InboundChannelAdapter(value = Source.OUTPUT, autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1")) public MessageSource timerMessageSource() { return () -> new GenericMessage<>(new SimpleDateFormat(this.options.getFormat()).format(new Date())); } diff --git a/spring-cloud-stream-samples/tap/src/main/java/config/TappingLoggingSink.java b/spring-cloud-stream-samples/tap/src/main/java/config/TappingLoggingSink.java index dc1436afd..9167feb09 100644 --- a/spring-cloud-stream-samples/tap/src/main/java/config/TappingLoggingSink.java +++ b/spring-cloud-stream-samples/tap/src/main/java/config/TappingLoggingSink.java @@ -18,25 +18,20 @@ package config; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import org.springframework.cloud.stream.annotation.EnableModule; -import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.Sink; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.messaging.MessageChannel; /** * @author Dave Syer * @author Marius Bogoevici */ -@EnableModule +@EnableModule(Sink.class) public class TappingLoggingSink { - + private static Logger logger = LoggerFactory.getLogger(TappingLoggingSink.class); - @Input - public MessageChannel input; - - @ServiceActivator(inputChannel="input") + @ServiceActivator(inputChannel = Sink.INPUT) public void loggerSink(Object payload) { logger.info("Received: " + payload); } diff --git a/spring-cloud-stream-samples/tap/src/main/java/demo/TapApplication.java b/spring-cloud-stream-samples/tap/src/main/java/demo/TapApplication.java index d3be7c678..c57894628 100644 --- a/spring-cloud-stream-samples/tap/src/main/java/demo/TapApplication.java +++ b/spring-cloud-stream-samples/tap/src/main/java/demo/TapApplication.java @@ -2,13 +2,11 @@ package demo; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.cloud.stream.annotation.EnableModule; import org.springframework.context.annotation.ComponentScan; import config.TappingLoggingSink; @SpringBootApplication -@EnableModule @ComponentScan(basePackageClasses= TappingLoggingSink.class) public class TapApplication { diff --git a/spring-cloud-stream-samples/transform/src/main/java/transform/LoggingTransformer.java b/spring-cloud-stream-samples/transform/src/main/java/transform/LoggingTransformer.java index 70e9a0822..ae36b0fdf 100644 --- a/spring-cloud-stream-samples/transform/src/main/java/transform/LoggingTransformer.java +++ b/spring-cloud-stream-samples/transform/src/main/java/transform/LoggingTransformer.java @@ -18,20 +18,16 @@ package transform; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.cloud.stream.annotation.EnableModule; -import org.springframework.context.annotation.Bean; +import org.springframework.cloud.stream.annotation.Processor; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.SubscribableChannel; /** * @author Dave Syer * */ -@EnableModule +@EnableModule(Processor.class) @ConfigurationProperties("module.logging") public class LoggingTransformer { @@ -50,17 +46,7 @@ public class LoggingTransformer { this.name = name; } - @Bean - public MessageChannel input() { - return new DirectChannel(); - } - - @Bean - public SubscribableChannel output() { - return new DirectChannel(); - } - - @ServiceActivator(inputChannel = "input", outputChannel = "output") + @ServiceActivator(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) public Object transform(Object payload) { logger.info("Transformed by " + this.name + ": " + payload); return payload; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java index e00c651b9..b2450819c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java @@ -47,8 +47,10 @@ import org.springframework.integration.annotation.MessageEndpoint; @Configuration @MessageEndpoint @Import({RedisServiceConfiguration.class, RabbitServiceConfiguration.class, - ChannelBindingAdapterConfiguration.class, CodecConfiguration.class, LifecycleConfiguration.class, - AggregateBuilderConfiguration.class, EnableModuleConfiguration.class}) + ChannelBindingAdapterConfiguration.class, CodecConfiguration.class, LifecycleConfiguration.class, + AggregateBuilderConfiguration.class, EnableModuleConfiguration.class}) public @interface EnableModule { + Class[] value() default {}; + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java index 586cce5aa..d3da85ef6 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java @@ -23,16 +23,16 @@ 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; /** - * Indicates that an input channel will be created and injected by the framework. + * Indicates that an input channel will be created by the framework. * * @author Dave Syer * @author Marius Bogoevici */ +@Qualifier @Target({ ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE }) @Retention(RetentionPolicy.RUNTIME) @@ -40,4 +40,6 @@ import org.springframework.beans.factory.annotation.Qualifier; @Documented public @interface Input { + String value() default ""; + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/ModuleChannels.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/ModuleChannels.java new file mode 100644 index 000000000..754f7730b --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/ModuleChannels.java @@ -0,0 +1,44 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +import org.springframework.beans.factory.annotation.Qualifier; + +/** + * Indicates an instance of a channels interface containing methods returning named + * message channels. + * + * @author Dave Syer + */ + +@Qualifier +@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE }) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +public @interface ModuleChannels { + + Class value(); + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java index a9d62e26a..a605a32c7 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java @@ -23,20 +23,22 @@ 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; /** - * Indicates that an output channel will be created and injected by the framework. + * Indicates that an output channel will be created by the framework. * * @author Dave Syer * @author Marius Bogoevici */ +@Qualifier @Target({ ElementType.FIELD, ElementType.METHOD}) @Retention(RetentionPolicy.RUNTIME) @Inherited @Documented public @interface Output { + String value() default ""; + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Processor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Processor.java new file mode 100644 index 000000000..d24c7aa41 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Processor.java @@ -0,0 +1,5 @@ +package org.springframework.cloud.stream.annotation; + +public interface Processor extends Source, Sink { + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Sink.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Sink.java new file mode 100644 index 000000000..37774ee71 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Sink.java @@ -0,0 +1,12 @@ +package org.springframework.cloud.stream.annotation; + +import org.springframework.messaging.SubscribableChannel; + +public interface Sink { + + public static String INPUT = "input"; + + @Input(Sink.INPUT) + SubscribableChannel input(); + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Source.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Source.java new file mode 100644 index 000000000..419924fee --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Source.java @@ -0,0 +1,15 @@ +package org.springframework.cloud.stream.annotation; + +import org.springframework.messaging.MessageChannel; + +public interface Source { + + public static String OUTPUT = "output"; + + @Output(Source.OUTPUT) + MessageChannel output(); + +} + + + \ No newline at end of file diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java index b1af42fc1..59c75c71f 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java @@ -15,17 +15,14 @@ */ package org.springframework.cloud.stream.config; -import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.LinkedHashSet; -import java.util.Map; import java.util.Properties; 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; @@ -34,7 +31,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.AbstractBeanDefinition; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.cloud.stream.adapter.ChannelBindingAdapter; import org.springframework.cloud.stream.adapter.ChannelLocator; import org.springframework.cloud.stream.adapter.InputChannelBinding; @@ -42,17 +38,12 @@ import org.springframework.cloud.stream.adapter.OutputChannelBinding; import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.Output; import org.springframework.cloud.stream.endpoint.ChannelsEndpoint; -import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; import org.springframework.xd.dirt.integration.bus.MessageBus; import org.springframework.xd.dirt.integration.bus.MessageBusAwareRouterBeanPostProcessor; -import org.springframework.xd.dirt.integration.bus.serializer.MultiTypeCodec; -import org.springframework.xd.dirt.integration.bus.serializer.kryo.FileKryoRegistrar; -import org.springframework.xd.dirt.integration.bus.serializer.kryo.KryoRegistrar; -import org.springframework.xd.dirt.integration.bus.serializer.kryo.PojoCodec; /** * @author Dave Syer diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnableModuleConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnableModuleConfiguration.java index 37df97e5c..ea77f6665 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnableModuleConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnableModuleConfiguration.java @@ -16,18 +16,48 @@ package org.springframework.cloud.stream.config; -import org.springframework.context.annotation.Bean; +import java.util.ArrayList; +import java.util.List; + +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.cloud.stream.annotation.EnableModule; +import org.springframework.cloud.stream.utils.MessageChannelBeanDefinitionRegistryUtils; import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; +import org.springframework.core.type.AnnotationMetadata; +import org.springframework.util.ClassUtils; +import org.springframework.util.MultiValueMap; /** * @author Marius Bogoevici + * @author Dave Syer */ @Configuration -public class EnableModuleConfiguration { +public class EnableModuleConfiguration implements ImportBeanDefinitionRegistrar { - @Bean - public ModulePostProcessor modulePostProcessor() { - return new ModulePostProcessor(); + @Override + public void registerBeanDefinitions(AnnotationMetadata metadata, + BeanDefinitionRegistry registry) { + MultiValueMap attributes = metadata.getAllAnnotationAttributes( + EnableModule.class.getName(), false); + for (Class type : collectClasses(attributes.get("value"))) { + MessageChannelBeanDefinitionRegistryUtils.registerChannelBeanDefinitions(type, registry); + MessageChannelBeanDefinitionRegistryUtils.registerChannelsQualifiedBeanDefinitions( + ClassUtils.resolveClassName(metadata.getClassName(), null), type, + registry); + } + } + + private List> collectClasses(List list) { + ArrayList> result = new ArrayList>(); + for (Object object : list) { + for (Object value : (Object[]) object) { + if (value instanceof Class && void.class != value) { + result.add((Class) value); + } + } + } + return result; } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ModulePostProcessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ModulePostProcessor.java deleted file mode 100644 index dabbbe83e..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ModulePostProcessor.java +++ /dev/null @@ -1,100 +0,0 @@ -/* - * Copyright 2015 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.config; - -import 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.BeanDefinitionRegistry; -import org.springframework.beans.factory.support.BeanDefinitionRegistryPostProcessor; -import org.springframework.cloud.stream.annotation.EnableModule; -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.Output; -import org.springframework.cloud.stream.utils.BeanDefinitionRegistryUtils; -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(final 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)) { - BeanDefinitionRegistryUtils.registerInputChannelBeanDefinition(field.getName(), registry); - } - if (field.isAnnotationPresent(Output.class)) { - BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition(field.getName(), registry); - } - } - }); - } - } catch (ClassNotFoundException e) { - throw new IllegalStateException(e); - } - } - } - - @Override - public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { - } - - @Override - public Object postProcessBeforeInitialization(final 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-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/BeanDefinitionRegistryUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/BeanDefinitionRegistryUtils.java deleted file mode 100644 index 13d23f9cd..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/BeanDefinitionRegistryUtils.java +++ /dev/null @@ -1,49 +0,0 @@ -/* - * Copyright 2015 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.utils; - -import java.lang.annotation.Annotation; - -import org.springframework.beans.factory.support.AutowireCandidateQualifier; -import org.springframework.beans.factory.support.BeanDefinitionRegistry; -import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.Output; -import org.springframework.cloud.stream.config.DirectChannelFactoryBean; - -/** - * Utility class for registering bean definitions. - * - * @author Marius Bogoevici - */ -public abstract class BeanDefinitionRegistryUtils { - - public static void registerInputChannelBeanDefinition(String name, BeanDefinitionRegistry registry) { - registerChannelBeanDefinition(Input.class, name, registry); - } - - public static void registerOutputChannelBeanDefinition(String name, BeanDefinitionRegistry registry) { - registerChannelBeanDefinition(Output.class, name, registry); - } - - private static void registerChannelBeanDefinition(Class qualifier, String name, BeanDefinitionRegistry registry) { - RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(DirectChannelFactoryBean.class); - rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(qualifier)); - registry.registerBeanDefinition(name, rootBeanDefinition); - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/MessageChannelBeanDefinitionRegistryUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/MessageChannelBeanDefinitionRegistryUtils.java new file mode 100644 index 000000000..92dcbe259 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/MessageChannelBeanDefinitionRegistryUtils.java @@ -0,0 +1,179 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.utils; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.Map; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.support.AutowireCandidateQualifier; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.ModuleChannels; +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.cloud.stream.config.DirectChannelFactoryBean; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.ReflectionUtils.MethodCallback; +import org.springframework.util.StringUtils; + +/** + * Utility class for registering bean definitions for message channels. + * + * @author Marius Bogoevici + * @author Dave Syer + */ +public abstract class MessageChannelBeanDefinitionRegistryUtils { + + public static void registerInputChannelBeanDefinition(String name, + BeanDefinitionRegistry registry) { + registerChannelBeanDefinition(Input.class, name, registry); + } + + public static void registerOutputChannelBeanDefinition(String name, + BeanDefinitionRegistry registry) { + registerChannelBeanDefinition(Output.class, name, registry); + } + + private static void registerChannelBeanDefinition( + Class qualifier, String name, + BeanDefinitionRegistry registry) { + RootBeanDefinition rootBeanDefinition = new RootBeanDefinition( + DirectChannelFactoryBean.class); + rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(qualifier)); + registry.registerBeanDefinition(name, rootBeanDefinition); + } + + public static void registerChannelBeanDefinitions(Class type, + final BeanDefinitionRegistry registry) { + ReflectionUtils.doWithMethods(type, new MethodCallback() { + @Override + public void doWith(Method method) throws IllegalArgumentException, + IllegalAccessException { + Input input = AnnotationUtils.findAnnotation(method, Input.class); + if (input != null) { + String name = getName(input, method); + registerInputChannelBeanDefinition(name, registry); + } + Output output = AnnotationUtils.findAnnotation(method, Output.class); + if (output != null) { + String name = getName(output, method); + registerOutputChannelBeanDefinition(name, registry); + } + } + + }); + } + + public static void registerChannelsQualifiedBeanDefinitions(Class parent, Class type, + final BeanDefinitionRegistry registry) { + + if (type.isInterface()) { + RootBeanDefinition rootBeanDefinition = new RootBeanDefinition( + ChannelProxyFactory.class); + rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(ModuleChannels.class, parent)); + rootBeanDefinition.getConstructorArgumentValues().addGenericArgumentValue( + type); + registry.registerBeanDefinition(type.getName(), rootBeanDefinition); + } + else { + RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(type); + rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(ModuleChannels.class, parent)); + registry.registerBeanDefinition(type.getName(), rootBeanDefinition); + } + } + + private static String getName(Annotation annotation, Method method) { + Map attrs = AnnotationUtils.getAnnotationAttributes(annotation, + false); + if (attrs.containsKey("value") && StringUtils.hasText((CharSequence) attrs.get("value"))) { + return (String) attrs.get("value"); + } + return method.getName(); + } + + static class ChannelProxyFactory implements MethodInterceptor, + FactoryBean, BeanFactoryAware { + + private Class type; + + private Object value = null; + + private BeanFactory beanFactory; + + public ChannelProxyFactory(Class type) { + this.type = type; + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + } + + @Override + public Object invoke(MethodInvocation invocation) throws Throwable { + Method method = invocation.getMethod(); + if (MessageChannel.class.isAssignableFrom(method.getReturnType())) { + Input input = AnnotationUtils.findAnnotation(method, Input.class); + if (input != null) { + String name = getName(input, method); + return this.beanFactory.getBean(name); + } + Output output = AnnotationUtils.findAnnotation(method, Output.class); + if (output != null) { + String name = getName(output, method); + return this.beanFactory.getBean(name); + } + } + return null; + } + + @Override + public Object getObject() throws Exception { + if (this.value == null) { + this.value = create(); + } + return this.value; + } + + private Object create() { + ProxyFactory factory = new ProxyFactory(this.type, this); + return factory.getProxy(); + } + + @Override + public Class getObjectType() { + return this.type; + } + + @Override + public boolean isSingleton() { + return true; + } + + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfigurationTests.java index ce8a279b1..7dbd2e862 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfigurationTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfigurationTests.java @@ -33,11 +33,10 @@ import org.springframework.cloud.stream.adapter.ChannelBinding; import org.springframework.cloud.stream.adapter.ChannelBindingAdapter; import org.springframework.cloud.stream.adapter.OutputChannelBinding; import org.springframework.cloud.stream.config.ChannelBindingAdapterConfigurationTests.Empty; -import org.springframework.cloud.stream.utils.BeanDefinitionRegistryUtils; +import org.springframework.cloud.stream.utils.MessageChannelBeanDefinitionRegistryUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; -import org.springframework.integration.channel.DirectChannel; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.annotation.DirtiesContext.ClassMode; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -69,7 +68,7 @@ public class ChannelBindingAdapterConfigurationTests { @Test public void oneOutput() throws Exception { - BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context); + MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context); refresh(); Collection channels = this.adapter.getChannelsMetadata().getOutputChannels(); assertEquals(1, channels.size()); @@ -88,7 +87,7 @@ public class ChannelBindingAdapterConfigurationTests { @Test public void oneOutputTopic() throws Exception { - BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.topic:", context); + MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.topic:", context); refresh(); Collection channels = this.adapter.getChannelsMetadata().getOutputChannels(); assertEquals(1, channels.size()); @@ -98,8 +97,8 @@ public class ChannelBindingAdapterConfigurationTests { @Test public void twoOutputsWithQueue() throws Exception { - BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context); - BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.queue:foo", context); + MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context); + MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.queue:foo", context); refresh(); Collection channels = this.adapter.getChannelsMetadata().getOutputChannels(); List names = getChannelNames(channels); @@ -119,7 +118,7 @@ public class ChannelBindingAdapterConfigurationTests { @Test public void overrideNaturalOutputChannelName() throws Exception { this.module.setOutputChannelName("bar"); - BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.queue:foo", context); + MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.queue:foo", context); refresh(); Collection channels = this.adapter.getChannelsMetadata().getOutputChannels(); assertEquals(1, channels.size()); @@ -131,7 +130,7 @@ public class ChannelBindingAdapterConfigurationTests { @Test public void overrideNaturalOutputChannelNamedQueueWithTopic() throws Exception { this.module.setOutputChannelName("queue:bar"); - BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.topic:foo", context); + MessageChannelBeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.topic:foo", context); refresh(); Collection channels = this.adapter.getChannelsMetadata().getOutputChannels(); assertEquals(1, channels.size());