From d7b8335224df4c54e5863cd0642a4e5d9c175b6e Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Thu, 28 May 2015 17:18:51 +0100 Subject: [PATCH] Add back missing code --- .gitignore | 3 +- roadmap.md | 6 + .../bus/runner/adapter/MessageBusAdapter.java | 64 ++++++---- .../MessageBusAdapterConfiguration.java | 3 - ...oduleOptionsPropertySourceInitializer.java | 117 ++++++++++++++++++ 5 files changed, 166 insertions(+), 27 deletions(-) create mode 100644 spring-xd-runner/src/main/java/org/springframework/bus/xd/bootstrap/ModuleOptionsPropertySourceInitializer.java diff --git a/.gitignore b/.gitignore index 41884cf75..166699fb5 100644 --- a/.gitignore +++ b/.gitignore @@ -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 diff --git a/roadmap.md b/roadmap.md index 192002590..fb2dbffcf 100644 --- a/roadmap.md +++ b/roadmap.md @@ -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. diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java index 48f7ec525..95191f17c 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java @@ -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 outputChannels = new LinkedHashMap(); + private Map inputChannels = new LinkedHashMap(); 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 outputChannels) { + this.outputChannels = outputChannels; + } + + public void setInputChannels(Map 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); } } diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java index 845e03b27..971bb60f9 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java @@ -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 { diff --git a/spring-xd-runner/src/main/java/org/springframework/bus/xd/bootstrap/ModuleOptionsPropertySourceInitializer.java b/spring-xd-runner/src/main/java/org/springframework/bus/xd/bootstrap/ModuleOptionsPropertySourceInitializer.java new file mode 100644 index 000000000..8e862a8c6 --- /dev/null +++ b/spring-xd-runner/src/main/java/org/springframework/bus/xd/bootstrap/ModuleOptionsPropertySourceInitializer.java @@ -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 { + + @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 map = new LinkedHashMap(); + 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 delegates = new ArrayList(); + 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(); + } + } + +}