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

Add back missing code

Browse Source
This commit is contained in:
Dave Syer
2015-05-28 17:18:51 +01:00
parent f58703acdb
commit d7b8335224
5 changed files with 166 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -7,7 +7,6 @@ asciidoctor.css
target/
bin/
_site/
xd
.classpath
.project
.settings
@@ -17,4 +16,4 @@ xd
*.iml
.idea
.factorypath
spring-xd-module-runner-sample/xd
spring-xd-samples/*/xd

6
roadmap.md
View File

@@ -71,6 +71,8 @@ To be deployable as an XD module in a "traditional" way you need `/config/*.prop
- [ ] Support for multiple input and output channels
- [ ] Partitioning
- [ ] Support for pubsub as "primary" input/output (in addition to the existing queue semantics)
- [ ] Support for more than one `MessageBus` (e.g. local and redis) in the same app
@@ -83,6 +85,8 @@ To be deployable as an XD module in a "traditional" way you need `/config/*.prop
- [ ] Re-use existing XD analytics as libraries (possibly attempt merge with Spring Boot metrics)
- [ ] Support Spring Batch jobs as modules
## Barriers to Progress
The best plan for making progress, where we keep in sight the goal of eventually having Spring XD converge with this project, is to shadow Spring XD andtry and extract as much goodness from it as we can. The `MessageBus` is really the core concept and it is already largely split out.
@@ -107,6 +111,8 @@ The best plan for making progress, where we keep in sight the goal of eventually
- [ ] `XdHeaders` (e.g. for history)
- [ ] `BusUtils` (e.g. to construct external channel names)
- [ ] There is a curator dependency in Spring XD that can't be shaken off.
- [ ] Spring XD plugins provide a rich set of lifecycle hooks, but those would not all be needed and are an awkward mismatch with a "pure-play" Spring Boot approach, where the application is either running or not.

64
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java
View File

@@ -58,8 +58,9 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
private MessageBus messageBus;
private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private MessageChannel outputChannel;
private MessageChannel inputChannel;
private Map<String, OutputChannelSpec> outputChannels = new LinkedHashMap<String, OutputChannelSpec>();
private Map<String, InputChannelSpec> inputChannels = new LinkedHashMap<String, InputChannelSpec>();
private boolean running = false;
@@ -92,11 +93,25 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
}
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
if (outputChannel != null) {
String name = module.getOutputChannelName();
this.outputChannels.put(name, new OutputChannelSpec(name, outputChannel));
}
}
public void setInputChannel(MessageChannel inputChannel) {
this.inputChannel = inputChannel;
if (inputChannel != null) {
String name = module.getInputChannelName();
this.inputChannels.put(name, new InputChannelSpec(name, inputChannel));
}
}
public void setOutputChannels(Map<String, OutputChannelSpec> outputChannels) {
this.outputChannels = outputChannels;
}
public void setInputChannels(Map<String, InputChannelSpec> inputChannels) {
this.inputChannels = inputChannels;
}
@Override
@@ -137,13 +152,15 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
}
protected final void unbindChannels() {
if (inputChannel != null) {
messageBus.unbindConsumers(module.getInputChannelName());
for (String name : inputChannels.keySet()) {
messageBus.unbindConsumers(name);
}
if (outputChannel != null) {
messageBus.unbindProducers(module.getOutputChannelName());
String tapChannelName = module.getTapChannelName();
messageBus.unbindProducers(tapChannelName);
for (String name : outputChannels.keySet()) {
messageBus.unbindProducers(name);
if (outputChannels.get(name).isTapped()) {
String tapChannelName = outputChannels.get(name).getTapChannelName();
messageBus.unbindProducers(tapChannelName);
}
}
}
@@ -152,22 +169,25 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
if (trackHistory) {
// TODO: addHistoryTag();
}
if (outputChannel != null) {
bindMessageProducer(outputChannel, module.getOutputChannelName(),
module.getProducerProperties());
String tapChannelName = module.getTapChannelName();
// tappableChannels.put(tapChannelName, outputChannel);
// if (isTapActive(tapChannelName)) {
createAndBindTapChannel(tapChannelName, outputChannel);
// }
for (String name : outputChannels.keySet()) {
OutputChannelSpec spec = outputChannels.get(name);
MessageChannel outputChannel = spec.getMessageChannel();
bindMessageProducer(outputChannel, name, module.getProducerProperties());
if (spec.isTapped()) {
String tapChannelName = spec.getTapChannelName();
// tappableChannels.put(tapChannelName, outputChannel);
// if (isTapActive(tapChannelName)) {
createAndBindTapChannel(tapChannelName, outputChannel);
// }
}
if (trackHistory) {
track(outputChannel, historyProperties);
}
}
if (inputChannel != null) {
bindMessageConsumer(inputChannel, module.getInputChannelName(),
module.getConsumerProperties());
if (trackHistory && outputChannel==null) {
for (String name : inputChannels.keySet()) {
MessageChannel inputChannel = inputChannels.get(name).getMessageChannel();
bindMessageConsumer(inputChannel, name, module.getConsumerProperties());
if (trackHistory && outputChannels.size() != 1) {
track(inputChannel, historyProperties);
}
}

3
spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java
View File

@@ -19,12 +19,10 @@ import java.util.Properties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.bus.runner.adapter.MessageBusAdapter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.ImportResource;
import org.springframework.messaging.MessageChannel;
import org.springframework.xd.dirt.integration.bus.MessageBus;
@@ -35,7 +33,6 @@ import org.springframework.xd.dirt.integration.bus.MessageBusAwareRouterBeanPost
*
*/
@Configuration
@Import(PropertyPlaceholderAutoConfiguration.class)
@ImportResource("classpath*:/META-INF/spring-xd/bus/codec.xml")
@EnableConfigurationProperties(MessageBusProperties.class)
public class MessageBusAdapterConfiguration {

117
spring-xd-runner/src/main/java/org/springframework/bus/xd/bootstrap/ModuleOptionsPropertySourceInitializer.java Normal file
View File

@@ -0,0 +1,117 @@
/*
* 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.bus.xd.bootstrap;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.bus.runner.config.MessageBusProperties;
import org.springframework.context.ApplicationContextInitializer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.Environment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.xd.dirt.plugins.job.JobPluginMetadataResolver;
import org.springframework.xd.dirt.plugins.stream.ModuleTypeConversionPluginMetadataResolver;
import org.springframework.xd.module.ModuleDefinition;
import org.springframework.xd.module.ModuleDefinitions;
import org.springframework.xd.module.ModuleType;
import org.springframework.xd.module.options.DefaultModuleOptionsMetadataResolver;
import org.springframework.xd.module.options.DelegatingModuleOptionsMetadataResolver;
import org.springframework.xd.module.options.EnvironmentAwareModuleOptionsMetadataResolver;
import org.springframework.xd.module.options.ModuleOption;
import org.springframework.xd.module.options.ModuleOptionsMetadata;
import org.springframework.xd.module.options.ModuleOptionsMetadataResolver;
/**
* Initialize the application context with default values for the module options.
*
* @author Dave Syer
*
*/
@Configuration
@EnableConfigurationProperties(MessageBusProperties.class)
@Order(Ordered.HIGHEST_PRECEDENCE + 10)
public class ModuleOptionsPropertySourceInitializer implements
ApplicationContextInitializer<ConfigurableApplicationContext> {
@Autowired
private MessageBusProperties module = new MessageBusProperties();
@Autowired(required=false)
private EnvironmentAwareModuleOptionsMetadataResolver wrapper;
@Override
public void initialize(ConfigurableApplicationContext applicationContext) {
ConfigurableEnvironment environment = applicationContext.getEnvironment();
ModuleOptionsMetadataResolver resolver = moduleOptionsMetadataResolver(environment);
ModuleOptionsMetadata resolved = resolver.resolve(getModuleDefinition());
Map<String, Object> map = new LinkedHashMap<String, Object>();
for (ModuleOption option : resolved) {
map.put(option.getName(), option.getDefaultValue());
}
insert(environment, new MapPropertySource("moduleDefaults", map));
}
private ModuleDefinition getModuleDefinition() {
return ModuleDefinitions.simple(module.getName(),
ModuleType.valueOf(module.getType()), "classpath:");
}
private void insert(ConfigurableEnvironment environment, MapPropertySource source) {
environment.getPropertySources().addLast(source);
}
private ModuleOptionsMetadataResolver moduleOptionsMetadataResolver(Environment environment) {
List<ModuleOptionsMetadataResolver> delegates = new ArrayList<ModuleOptionsMetadataResolver>();
delegates.add(defaultResolver());
delegates.add(new ModuleTypeConversionPluginMetadataResolver());
delegates.add(new JobPluginMetadataResolver());
DelegatingModuleOptionsMetadataResolver delegatingResolver = new DelegatingModuleOptionsMetadataResolver();
delegatingResolver.setDelegates(delegates);
ModuleOptionsMetadataResolver resolver = delegatingResolver;
if (wrapper!=null) {
wrapper.setDelegate(delegatingResolver);
resolver = wrapper;
}
return resolver;
}
@Bean
// TODO: allow override of this
public DefaultModuleOptionsMetadataResolver defaultResolver() {
return new DefaultModuleOptionsMetadataResolver();
}
@ConditionalOnExpression("'${xd.module.config.location:${xd.config.home:}}'!=''")
protected static class EnvironmentAwareModuleOptionsMetadataResolverConfiguration {
@Bean
public EnvironmentAwareModuleOptionsMetadataResolver environmentAwareModuleOptionsMetadataResolver() {
return new EnvironmentAwareModuleOptionsMetadataResolver();
}
}
}