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

Identify bindings with annotations, not names

Browse Source 
Refactor support for channel beans

* create utility methods for registering input/output channel beans
* use utility methods within tests

Remove source-xml and rely on the @Input/@Output model entirely
This commit is contained in:
Marius Bogoevici
2015-07-10 14:18:15 -04:00
committed by Mark Fisher
parent 23e4cbc7ac
commit a68321e8d5
14 changed files with 86 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,11 +31,16 @@ import org.springframework.aop.target.LazyInitTargetSource;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.ListableBeanFactory;
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.streams.adapter.ChannelBindingAdapter;
import org.springframework.cloud.streams.adapter.ChannelLocator;
import org.springframework.cloud.streams.adapter.InputChannelBinding;
import org.springframework.cloud.streams.adapter.OutputChannelBinding;
import org.springframework.cloud.streams.annotation.Input;
import org.springframework.cloud.streams.annotation.Output;
import org.springframework.cloud.streams.endpoint.ChannelsEndpoint;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
@@ -62,7 +67,7 @@ public class ChannelBindingAdapterConfiguration {
private ChannelBindingProperties module;
@Autowired
private ListableBeanFactory beanFactory;
private ConfigurableListableBeanFactory beanFactory;
private ChannelLocator channelLocator;
@@ -92,10 +97,13 @@ public class ChannelBindingAdapterConfiguration {
}
protected Collection<OutputChannelBinding> getOutputChannels() {
Set<OutputChannelBinding> channels = new LinkedHashSet<OutputChannelBinding>();
Set<OutputChannelBinding> channels = new LinkedHashSet<>();
String[] names = this.beanFactory.getBeanNamesForType(MessageChannel.class);
for (String name : names) {
if (name.startsWith("output")) {
BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(name);
// for now, just assume that the beans are at least AbstractBeanDefinition
if (beanDefinition instanceof AbstractBeanDefinition
&& ((AbstractBeanDefinition)beanDefinition).getQualifier(Output.class.getName()) != null) {
channels.add(new OutputChannelBinding(name));
}
}
@@ -103,10 +111,13 @@ public class ChannelBindingAdapterConfiguration {
}
protected Collection<InputChannelBinding> getInputChannels() {
Set<InputChannelBinding> channels = new LinkedHashSet<InputChannelBinding>();
Set<InputChannelBinding> channels = new LinkedHashSet<>();
String[] names = this.beanFactory.getBeanNamesForType(MessageChannel.class);
for (String name : names) {
if (name.startsWith("input")) {
BeanDefinition beanDefinition = this.beanFactory.getBeanDefinition(name);
// for now, just assume that the beans are at least AbstractBeanDefinition
if (beanDefinition instanceof AbstractBeanDefinition
&& ((AbstractBeanDefinition)beanDefinition).getQualifier(Input.class.getName()) != null) {
channels.add(new InputChannelBinding(name));
}
}

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

@@ -22,13 +22,12 @@ 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.cloud.streams.utils.BeanDefinitionRegistryUtils;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.AnnotationUtils;
@@ -59,14 +58,10 @@ public class ModulePostProcessor implements BeanDefinitionRegistryPostProcessor,
@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);
BeanDefinitionRegistryUtils.registerInputChannelBeanDefinition(field.getName(), registry);
}
if (field.isAnnotationPresent(Output.class)) {
RootBeanDefinition rootBeanDefinition = new RootBeanDefinition(DirectChannelFactoryBean.class);
rootBeanDefinition.addQualifier(new AutowireCandidateQualifier(Output.class));
registry.registerBeanDefinition(field.getName(), rootBeanDefinition);
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition(field.getName(), registry);
}
}
});

49
spring-cloud-streams/src/main/java/org/springframework/cloud/streams/utils/BeanDefinitionRegistryUtils.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* 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.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.streams.annotation.Input;
import org.springframework.cloud.streams.annotation.Output;
import org.springframework.cloud.streams.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);
}
}

13
spring-cloud-streams/src/test/java/org/springframework/cloud/streams/config/ChannelBindingAdapterConfigurationTests.java
View File

@@ -33,6 +33,7 @@ import org.springframework.cloud.streams.adapter.ChannelBinding;
import org.springframework.cloud.streams.adapter.ChannelBindingAdapter;
import org.springframework.cloud.streams.adapter.OutputChannelBinding;
import org.springframework.cloud.streams.config.ChannelBindingAdapterConfigurationTests.Empty;
import org.springframework.cloud.streams.utils.BeanDefinitionRegistryUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
@@ -68,7 +69,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void oneOutput() throws Exception {
this.context.registerSingleton("output", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
assertEquals(1, channels.size());
@@ -87,7 +88,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void oneOutputTopic() throws Exception {
this.context.registerSingleton("output.topic:", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.topic:", context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
assertEquals(1, channels.size());
@@ -97,8 +98,8 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void twoOutputsWithQueue() throws Exception {
this.context.registerSingleton("output", new DirectChannel());
this.context.registerSingleton("output.queue:foo", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context);
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.queue:foo", context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
List<String> names = getChannelNames(channels);
@@ -118,7 +119,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void overrideNaturalOutputChannelName() throws Exception {
this.module.setOutputChannelName("bar");
this.context.registerSingleton("output.queue:foo", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.queue:foo", context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
assertEquals(1, channels.size());
@@ -130,7 +131,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void overrideNaturalOutputChannelNamedQueueWithTopic() throws Exception {
this.module.setOutputChannelName("queue:bar");
this.context.registerSingleton("output.topic:foo", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.topic:foo", context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata().getOutputChannels();
assertEquals(1, channels.size());

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

@@ -35,6 +35,7 @@ import org.springframework.cloud.streams.adapter.ChannelBindingAdapter;
import org.springframework.cloud.streams.adapter.InputChannelBinding;
import org.springframework.cloud.streams.adapter.OutputChannelBinding;
import org.springframework.cloud.streams.config.ChannelBindingAdapterConfiguration;
import org.springframework.cloud.streams.utils.BeanDefinitionRegistryUtils;
import org.springframework.cloud.streams.xd.ChannelBindingAdapterConfigurationTests.Empty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -71,6 +72,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void oneOutput() throws Exception {
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", context);
this.context.registerSingleton("output", new DirectChannel());
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata()
@@ -83,7 +85,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void oneInput() throws Exception {
this.context.registerSingleton("input", new DirectChannel());
BeanDefinitionRegistryUtils.registerInputChannelBeanDefinition("input",context);
refresh();
Collection<InputChannelBinding> channels = this.adapter.getChannelsMetadata()
.getInputChannels();
@@ -104,7 +106,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void oneOutputTopic() throws Exception {
this.context.registerSingleton("output.topic:", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.topic:", this.context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata()
.getOutputChannels();
@@ -119,7 +121,7 @@ public class ChannelBindingAdapterConfigurationTests {
this.module.setGroup("mine");
this.module.setName("foo");
this.module.setIndex(2);
this.context.registerSingleton("output.topic:", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.topic:", this.context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata()
.getOutputChannels();
@@ -131,7 +133,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void oneInputTopic() throws Exception {
this.context.registerSingleton("input.topic:", new DirectChannel());
BeanDefinitionRegistryUtils.registerInputChannelBeanDefinition("input.topic:", this.context);
refresh();
Collection<InputChannelBinding> channels = this.adapter.getChannelsMetadata()
.getInputChannels();
@@ -143,7 +145,7 @@ public class ChannelBindingAdapterConfigurationTests {
public void oneInputOverrideName() throws Exception {
this.module.setGroup("mine");
this.module.setIndex(2);
this.context.registerSingleton("input", new DirectChannel());
BeanDefinitionRegistryUtils.registerInputChannelBeanDefinition("input", this.context);
refresh();
Collection<InputChannelBinding> channels = this.adapter.getChannelsMetadata()
.getInputChannels();
@@ -153,8 +155,8 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void twoOutputsWithQueue() throws Exception {
this.context.registerSingleton("output", new DirectChannel());
this.context.registerSingleton("output.queue:foo", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output", this.context);
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.queue:foo", this.context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata()
.getOutputChannels();
@@ -175,7 +177,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void overrideNaturalOutputChannelName() throws Exception {
this.module.setOutputChannelName("bar");
this.context.registerSingleton("output.queue:foo", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.queue:foo", this.context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata()
.getOutputChannels();
@@ -189,7 +191,7 @@ public class ChannelBindingAdapterConfigurationTests {
@Test
public void overrideNaturalOutputChannelNamedQueueWithTopic() throws Exception {
this.module.setOutputChannelName("queue:bar");
this.context.registerSingleton("output.topic:foo", new DirectChannel());
BeanDefinitionRegistryUtils.registerOutputChannelBeanDefinition("output.topic:foo", this.context);
refresh();
Collection<OutputChannelBinding> channels = this.adapter.getChannelsMetadata()
.getOutputChannels();

1
spring-xd-samples/pom.xml
View File

@@ -19,7 +19,6 @@
<module>source</module>
<module>sink</module>
<module>tap</module>
<module>source-xml</module>
</modules>
<build>
<pluginManagement>

60
spring-xd-samples/source-xml/pom.xml
View File

@@ -1,60 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-xd-module-runner-sample-source-xml</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<name>spring-xd-module-runner-sample-source-xml</name>
<description>Demo project for Spring XD module</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-xd-samples</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<start-class>demo.ModuleApplication</start-class>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-xd-runner</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-messagebus-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>

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

@@ -1,19 +0,0 @@
package demo;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.streams.annotation.EnableModule;
import org.springframework.context.annotation.ImportResource;
import org.springframework.context.annotation.PropertySource;
@SpringBootApplication
@EnableModule
@ImportResource("classpath:/config/ticker.xml")
@PropertySource("classpath:/config/ticker.properties")
public class ModuleApplication {
public static void main(String[] args) throws InterruptedException {
new SpringApplicationBuilder().sources(ModuleApplication.class).run(args);
}
}

61
spring-xd-samples/source-xml/src/main/java/demo/TimeSourceOptionsMetadata.java
View File

@@ -1,61 +0,0 @@
/*
* Copyright 2013-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 demo;
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.spi.ModuleOption;
import org.springframework.xd.module.options.validation.DateFormat;
/**
* Describes options to the {@code time} source module.
*
* @author Eric Bottard
* @author Gary Russell
*/
@Mixin({ PeriodicTriggerMixin.class, MaxMessagesDefaultOneMixin.class })
public class TimeSourceOptionsMetadata {
private String format = "yyyy-MM-dd HH:mm:ss";
private int fixedDelay = 1;
@DateFormat
public String getFormat() {
return format;
}
@ModuleOption("how to render the current time, using SimpleDateFormat")
public void setFormat(String format) {
this.format = format;
}
public int getFixedDelay() {
return fixedDelay;
}
@ModuleOption("time delay between messages, expressed in TimeUnits (seconds by default)")
public void setFixedDelay(int fixedDelay) {
this.fixedDelay = fixedDelay;
}
}

1
spring-xd-samples/source-xml/src/main/resources/application.yml
View File

@@ -1 +0,0 @@
fixedDelay: 5

6
spring-xd-samples/source-xml/src/main/resources/bootstrap.yml
View File

@@ -1,6 +0,0 @@
---
spring:
cloud:
channels:
group: testtock
name: ticker

1
spring-xd-samples/source-xml/src/main/resources/config/ticker.properties
View File

@@ -1 +0,0 @@
options_class = demo.TimeSourceOptionsMetadata

22
spring-xd-samples/source-xml/src/main/resources/config/ticker.xml
View File

@@ -1,22 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<channel id="output"/>
<inbound-channel-adapter channel="output"
auto-startup="false"
expression="new java.text.SimpleDateFormat('${format}').format(new java.util.Date())">
<poller trigger="fixedDelayTrigger" />
</inbound-channel-adapter>
<beans:bean id="fixedDelayTrigger" class="org.springframework.scheduling.support.PeriodicTrigger">
<beans:constructor-arg value="${fixedDelay}" />
<beans:constructor-arg value="${timeUnit}" />
<beans:property name="initialDelay" value="${initialDelay} "/>
</beans:bean>
</beans:beans>

20
spring-xd-samples/source-xml/src/test/java/demo/ModuleApplicationTests.java
View File

@@ -1,20 +0,0 @@
package demo;
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.boot.test.SpringApplicationConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = ModuleApplication.class)
@WebAppConfiguration
@DirtiesContext
public class ModuleApplicationTests {
@Test
public void contextLoads() {
}
}