SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use @EnableModule(Channels.class) instead of field declarations

Browse Source 
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.
This commit is contained in:
Dave Syer
2015-07-16 15:10:41 +01:00
committed by Mark Fisher
parent 460bb140f2
commit c09b292d79
22 changed files with 390 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
docs/src/main/asciidoc/intro.adoc
View File

@@ -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<String> 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

11
spring-cloud-stream-samples/double/src/main/java/config/SinkModuleDefinition.java
View File

@@ -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);
}

10
spring-cloud-stream-samples/double/src/main/java/config/SourceModuleDefinition.java
View File

@@ -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<String> timerMessageSource() {
return () -> new GenericMessage<>(new SimpleDateFormat(this.format).format(new Date()));
}

11
spring-cloud-stream-samples/sink/src/main/java/sink/LogSink.java
View File

@@ -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);
}

25
spring-cloud-stream-samples/sink/src/test/java/demo/ModuleApplicationTests.java
View File

@@ -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());
}
}

10
spring-cloud-stream-samples/source/src/main/java/source/TimeSource.java
View File

@@ -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<String> timerMessageSource() {
return () -> new GenericMessage<>(new SimpleDateFormat(this.options.getFormat()).format(new Date()));
}

13
spring-cloud-stream-samples/tap/src/main/java/config/TappingLoggingSink.java
View File

@@ -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);
}

2
spring-cloud-stream-samples/tap/src/main/java/demo/TapApplication.java
View File

@@ -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 {

20
spring-cloud-stream-samples/transform/src/main/java/transform/LoggingTransformer.java
View File

@@ -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;

6
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableModule.java
View File

@@ -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 {};
}

6
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java
View File

@@ -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 "";
}

44
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/ModuleChannels.java Normal file
View File

@@ -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();
}

6
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java
View File

@@ -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 "";
}

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Processor.java Normal file
View File

@@ -0,0 +1,5 @@
package org.springframework.cloud.stream.annotation;
public interface Processor extends Source, Sink {
}

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Sink.java Normal file
View File

@@ -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();
}

15
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Source.java Normal file
View File

@@ -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();
}

9
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfiguration.java
View File

@@ -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

40
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/EnableModuleConfiguration.java
View File

@@ -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<String, Object> 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<Class<?>> collectClasses(List<Object> list) {
ArrayList<Class<?>> result = new ArrayList<Class<?>>();
for (Object object : list) {
for (Object value : (Object[]) object) {
if (value instanceof Class && void.class != value) {
result.add((Class<?>) value);
}
}
}
return result;
}
}

100
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ModulePostProcessor.java
View File

@@ -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;
}
}

49
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/BeanDefinitionRegistryUtils.java
View File

@@ -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<? extends Annotation> qualifier, String name, BeanDefinitionRegistry registry) {
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(DirectChannelFactoryBean.class);
rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(qualifier));
registry.registerBeanDefinition(name, rootBeanDefinition);
}
}

179
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/utils/MessageChannelBeanDefinitionRegistryUtils.java Normal file
View File

@@ -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<? extends Annotation> 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<String, Object> 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<Object>, 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;
}
}
}

15
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/ChannelBindingAdapterConfigurationTests.java
View File

@@ -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<OutputChannelBinding> 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<OutputChannelBinding> 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<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
List<String> 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<OutputChannelBinding> 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<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
assertEquals(1, channels.size());