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

Migrate some code from prototype

Browse Source
This commit is contained in:
Dave Syer
2015-05-28 16:42:00 +01:00
parent a288ed1fbb
commit f58703acdb
45 changed files with 1955 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
.gitignore vendored Normal file
View File

@@ -0,0 +1,20 @@
/application.yml
/application.properties
asciidoctor.css
*~
.#*
*#
target/
bin/
_site/
xd
.classpath
.project
.settings
.springBeans
.DS_Store
*.sw*
*.iml
.idea
.factorypath
spring-xd-module-runner-sample/xd

105
README.md
View File

@@ -1 +1,104 @@
# spring-bus
# Spring Integration: Messaging as a Microservice
This is an experimental project allowing a user to develop and run messaging microservices using Spring Integration and run them locally or in the cloud, or even on Spring XD. It also allows a user to develop and run an XD module locally. Just create `MessageChannels` "input" and/or "output" and add `@EnableMessageBus` 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):
```
@SpringBootApplication
@EnableMessageBus
@ComponentScan(basePackageClasses=ModuleDefinition.class)
public class ModuleApplication {
public static void main(String[] args) throws InterruptedException {
SpringApplication.run(ModuleApplication.class, args);
}
}
@Configuration
public class ModuleDefinition {
@Value("${format}")
private String format;
@Bean
public MessageChannel output() {
return new DirectChannel();
}
@Bean
@InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
public MessageSource<String> timerMessageSource() {
return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
}
}
```
The `bootstrap.yml` has the module group (a.k.a. stream name in XD), name and index, e.g.
```
---
spring:
bus:
group: testtock
name: ${spring.application.name:ticker}
index: 0 # source
```
To be deployable as an XD module in a "traditional" way you need `/config/*.properties` to point to any available Java config classes (via `base_packages` or `options_class`), or else you can put traditional XML configuration in `/config/*.xml`. You don't need those things to run as a consumer or producer to an existing XD system. There's an XML version of the same sample (a "timer" source).
## XD Module Samples
There are several samples, all running on the redis transport (so you need redis running locally to test them):
* `spring-xd-module-runner-sample-source` is a Java config version of the classic "timer" module from Spring XD. It has a "fixedDelay" option (in milliseconds) for the period between emitting messages.
* `spring-xd-module-runner-sample-sink` is a Java config version of the classic "log" module from Spring XD. It has no options (but some could easily be added), and just logs incoming messages at INFO level.
* `spring-xd-module-runner-sample-tap` is the same as the sink sample, except it is configured to tap the source sample output. When it is running it looks a lot like the sink, except that it only gets copies of the messages in the broker, and since it is a pub-sub subscriber, it only gets the messages sent since it started.
* `spring-xd-module-runner-sample-source-xml` is a copy of the classic "timer" module from Spring XD.
If you run the source and the sink and point them at the same redis instance (e.g. do nothing to get the one on localhost, or the one they are both bound to as a service on Cloud Foundry) then they will form a "stream" and start talking to each other. All the samples have friendly JMX and Actuator endpoints for inspecting what is going on in the system.
## Module or App
Code using this library can be deployed as a standalone app or as an XD module. In standalone mode you app will run happily as a service or in any PaaS (Cloud Foundry, Lattice, Heroku, Azure, etc.). Depending on whether your main aim is to develop an XD module and you just want to test it locally using the standalone mode, or if the ultimate goal is a standalone app, there are some things that you might do differently.
### Module Options
Module option (placeholders) default values can be set in `/config/*.properties` as per a normal XD module, and they can be overridden at runtime in standalone mode using standard Spring Boot configuration (e.g. `application.yml`). Because of the way XD likes to organize options, the default values can also be set as `option.*` in `bootstrap.yml` (in standalone mode) or as System properties (generally).
### Local Configuration
The `application.yml` and `bootstrap.yml` files are ignored by XD when deploying the module natively, so you can put whatever you like in there to control the app in standlone mode.
### Fat JAR
You can run in standalone mode from your IDE for testing. To run in production you can create an executable (or "fat") JAR using the standard Spring Boot tooling, but the executable JAR has a load of stuff in it that isn't needed if it's going to be deployed as an XD module. In that case you are better off with the normal JAR packaging provided by Maven or Gradle.
## Making Standalone Modules Talk to Each Other
The "group" and "index" are used to create physical endpoints in the external broker (e.g. `queue.<group>.<index>` in Redis), so a source (output only) has `index=0` (the default) and downstream modules have the same group but incremented index, with a sink module (input only) having the highest index. To listen to the output from an existing app, just use the same "group" name and an index 1 larger than the app before it in the chain. The index can be anything, as long as successive modules have consecutive values.
> Note: since the same naming conventions are used in XD, you can spy on or send messages to an existing XD stream by copying the stream name (to `spring.bus.group`) and knowing the index of the XD module you want to interact with.
## Taps
All output channels are also tapped by default so you can also attach a module to a pub-sub endpoint and listen to the tap if you know the module metadata (e.g. `topic.tap:stream:<name>.<group>.<index>` in Redis). To tap an existing output channel you just need to know its group, name and index, e.g.
```
spring:
bus:
group: tocktap
name: logger
index: 0
tap:
group: testtock
name: ticker
index: 0
```
The `spring.bus.tap` section tells the module runner which topic you want to subscribe to. It creates a new group (a tap can't be in the same group as the one it is tapping) and starts a new index count, in case anyone wants to listen downstream.

142
pom.xml Normal file
View File

@@ -0,0 +1,142 @@
<?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.bus</groupId>
<artifactId>spring-bus-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>pom</packaging>
<url>http://projects.spring.io/spring-xd/</url>
<organization>
<name>Pivotal Software, Inc.</name>
<url>http://www.spring.io</url>
</organization>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.2.3.RELEASE</version>
<relativePath /> <!-- lookup parent from repository -->
</parent>
<modules>
<module>spring-bus-core</module>
<module>spring-xd-runner</module>
<module>spring-xd-samples</module>
</modules>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-parent</artifactId>
<version>1.0.2.RELEASE</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<dependency>
<groupId>org.springframework.bus</groupId>
<artifactId>spring-bus-core</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.bus</groupId>
<artifactId>spring-xd-runner</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-dirt</artifactId>
<version>1.2.0.BUILD-SNAPSHOT</version>
<exclusions>
<exclusion>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-spark-streaming</artifactId>
</exclusion>
<exclusion>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-hadoop</artifactId>
</exclusion>
<exclusion>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-batch</artifactId>
</exclusion>
<exclusion>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-ui</artifactId>
</exclusion>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>zookeeper</artifactId>
<groupId>org.apache.zookeeper</groupId>
</exclusion>
<exclusion>
<artifactId>spring-security-ldap</artifactId>
<groupId>org.springframework.security</groupId>
</exclusion>
<exclusion>
<artifactId>spring-jdbc</artifactId>
<groupId>org.springframework</groupId>
</exclusion>
<exclusion>
<artifactId>spring-batch-integration</artifactId>
<groupId>org.springframework.batch</groupId>
</exclusion>
<exclusion>
<artifactId>spring-batch-admin-manager</artifactId>
<groupId>org.springframework.batch</groupId>
</exclusion>
<exclusion>
<artifactId>spring-data-mongodb</artifactId>
<groupId>org.springframework.data</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-messagebus-redis</artifactId>
<version>1.2.0.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-messagebus-rabbit</artifactId>
<version>1.2.0.BUILD-SNAPSHOT</version>
</dependency>
</dependencies>
</dependencyManagement>
<repositories>
<repository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>http://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
<repository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>http://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</repository>
</repositories>
<pluginRepositories>
<pluginRepository>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>http://repo.spring.io/snapshot</url>
<snapshots>
<enabled>true</enabled>
</snapshots>
</pluginRepository>
<pluginRepository>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>http://repo.spring.io/milestone</url>
<snapshots>
<enabled>false</enabled>
</snapshots>
</pluginRepository>
</pluginRepositories>
</project>

114
roadmap.md Normal file
View File

@@ -0,0 +1,114 @@
# Messaging Microservices
Messaging and asynchronous patterns are completely natural with microservices, but a lot of the start of the art material concentrates on HTTP, JSON and REST. This document is about "Message-driven Microservices" with Spring. It tracks the convergence of various ideas that are floating around in Spring Cloud, Spring Boot and Spring XD.
There is already a [`spring-xd-module-runner` project](https://github.com/dsyer/spring-xd-module-runner) where we started experimenting with allowing user to develop and run an XD module locally. We can extrapolate from there to a more flexible model that leads optionally to a deployable XD module, but can also be used to form more flexible structures than a simple "stream".
## Basic Programming Model
Just create `MessageChannels` "input" and/or "output" and add `@EnableMessageBus` and run your app as a Spring Boot app (single application context). You 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):
```
@SpringBootApplication
@EnableMessageBus
@ComponentScan(basePackageClasses=ModuleDefinition.class)
public class MessageBusApplication {
public static void main(String[] args) {
SpringApplication.run(MessageBusApplication.class, args);
}
}
@Configuration
public class ModuleDefinition {
@Value("${format}")
private String format;
@Bean
public MessageChannel output() {
return new DirectChannel();
}
@Bean
@InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
public MessageSource<String> timerMessageSource() {
return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
}
}
```
The `bootstrap.yml` has the module group (a.k.a. stream name), name and index, e.g.
```
---
spring:
bus:
group: testtock
name: ${spring.application.name:ticker}
index: 0 # source
```
## Richer Input and Output
A generic message-driven microservice can have more than 2 inputs and outputs. For example if it contains a router component, then it might want to send messages downstream to 2 (or more) completely different classes of consumers. The number and purpose of those downstream channels might even change at runtime.
Spring XD has the concept of a "stream" where modules pipe their output into the next module's input (like a UN\*X shell). This is effectively an opinionated, special-case implementation of the more general message-driven microservice, and in it should be possible to re-architect Spring XD so that it still adds value but builds on top of a framework that is more general.
Spring XD actually already supports arbitrary graphs of input and output via named channels. One of the aims of this project is to make that easier to program (or declare), and to make modules composable into a self-contained application.
> NOTE: For the sake of clarity, consider the example of an XD module that contains a router, sending each message to one of two destinations (for invalid or valid messages respectively). For XD purposes this module is a "sink" because it has input but no uniq output channel. The output is handled and redirected through the Bus using a `MessageChannelResolver` that is driven by naming convention (destinations start with "queue:" or "topic:" according to their pubsub semantics). Users can create streams that listen to those channels by using the same names.
## Deployment as an XD Module
To be deployable as an XD module in a "traditional" way you need `/config/*.properties` to point to any available Java config classes (via `base_packages` or `options_class`), or else you can put traditional XML configuration in `/config/*.xml`. You don't need those things to run as a consumer or producer to an existing XD system. The `spring-xd-module-runner` library uses a Spring Cloud bootstrap context to initialize the `Module` environment properties in a way that simulates being deployed in a "full" XD system.
## Backlog
- [ ] Support for multiple input and output channels
- [ ] 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
- [ ] Correlation and message tracing
- [ ] Extract `spring-xd-dirt` dependencies into a separate module
- [ ] Re-use existing XD modules as libraries
- [ ] Re-use existing XD analytics as libraries (possibly attempt merge with Spring Boot metrics)
## 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.
- [ ] Spring XD package names should eventually be removed as messaging is a lower level concern.
- [ ] Ditto configuration properties in `xd.*` should be moved to `spring.bus.*` (or something).
- [ ] We need the XML configuration from `/META-INF/spring-xd/bus/**` and `/META-INF/spring-xd/analytics` but
- [ ] There are no defaults for several properties in `xd.messagebus.*` so applications have to have a load of boilerplate configuration in `application.yml`
- [ ] The XML is in `spring-xd-dirt` which we don't want to depend on. Maybe it should be in the messagebus impementation jars?
- [x] Do we need the analytics configuration? It should at least be optional probably. Answer "no".
- [ ] The `spring-xd-dirt` library contains some of the primitives we might need, especially when building the bridge to create XD modules as apps. It would be best if they could be extracted into another library.
- [ ] `MessageBusAwareChannelResolver` and `MessageBusAwareRouterBeanPostProcessor` seem like they should be pulled out of dirt.
- [x] A `ModuleDefinition` is only needed to initialize options for an XD module (so not really needed for the general case). We can split the module options initializer outr into a separate module that you only need if you know you want to test an XD module with its native options meatdata.
- [ ] `XdHeaders` (e.g. for history)
- [ ] 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.
- [ ] Spring XD Module options are a prime example of the above. The `StreamPlugin` is responsible for setting up default values for "options" which are really nothing more than property sources in the `Environment` (at least from the point of view of the ideal developer experience). Some of the code from that was used in a Spring Cloud bootstrap context in the `spring-xd-module-runner` project, which puts it in the right part of the lifecycle. It might be better to try and wean XD off its native "options" and back onto a model based on vanilla Spring Boot and Spring Cloud features (`@ConfigurationProperties` and the bootstrap context for example).

62
spring-bus-core/pom.xml Normal file
View File

@@ -0,0 +1,62 @@
<?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.bus</groupId>
<artifactId>spring-bus-core</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<name>spring-bus-runner</name>
<description>Demo project for Spring Integration as apps</description>
<parent>
<groupId>org.springframework.bus</groupId>
<artifactId>spring-bus-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-dirt</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-messagebus-redis</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-messagebus-rabbit</artifactId>
<optional>true</optional>
</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-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

46
spring-bus-core/src/main/java/org/springframework/bus/runner/EnableMessageBus.java Normal file
View File

@@ -0,0 +1,46 @@
/*
* 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.runner;
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.bus.runner.config.LifecycleConfiguration;
import org.springframework.bus.runner.config.MessageBusAdapterConfiguration;
import org.springframework.bus.runner.config.RabbitServiceConfiguration;
import org.springframework.bus.runner.config.RedisServiceConfiguration;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
/**
* @author Dave Syer
*
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Configuration
@Import({ RedisServiceConfiguration.class, RabbitServiceConfiguration.class,
MessageBusAdapterConfiguration.class, LifecycleConfiguration.class })
public @interface EnableMessageBus {
}

43
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* 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.runner.adapter;
import org.springframework.messaging.MessageChannel;
/**
* @author Dave Syer
*
*/
public class InputChannelSpec {
private String name;
private MessageChannel channel;
public InputChannelSpec(String name, MessageChannel channel) {
this.name = name;
this.channel = channel;
}
public String getName() {
return name;
}
public MessageChannel getMessageChannel() {
return channel;
}
}

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

@@ -0,0 +1,267 @@
/*
* 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.runner.adapter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.bus.runner.config.MessageBusProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.interceptor.WireTap;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.xd.dirt.integration.bus.MessageBus;
import org.springframework.xd.dirt.integration.bus.XdHeaders;
/**
* @author Mark Fisher
* @author Dave Syer
*/
@ManagedResource
public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
private static Logger logger = LoggerFactory.getLogger(MessageBusAdapter.class);
private MessageBus messageBus;
private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private MessageChannel outputChannel;
private MessageChannel inputChannel;
private boolean running = false;
private final AtomicBoolean active = new AtomicBoolean(false);
private boolean trackHistory = false;
private MessageBusProperties module;
private ConfigurableApplicationContext applicationContext;
public MessageBusAdapter(MessageBusProperties module, MessageBus messageBus) {
this.module = module;
this.messageBus = messageBus;
}
@Override
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = (ConfigurableApplicationContext) applicationContext;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
}
public void setTrackHistory(boolean trackHistory) {
this.trackHistory = trackHistory;
}
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
public void setInputChannel(MessageChannel inputChannel) {
this.inputChannel = inputChannel;
}
@Override
@ManagedOperation
public void start() {
if (!running) {
// Start everything, but don't call ourselves
if (!active.get()) {
if (active.compareAndSet(false, true)) {
bindChannels();
applicationContext.start();
active.set(false);
}
}
}
running = true;
}
@Override
@ManagedOperation
public void stop() {
if (running) {
if (!active.get()) {
if (active.compareAndSet(false, true)) {
unbindChannels();
applicationContext.stop();
active.set(false);
}
}
}
running = false;
}
@Override
@ManagedAttribute
public boolean isRunning() {
return running && applicationContext.isRunning();
}
protected final void unbindChannels() {
if (inputChannel != null) {
messageBus.unbindConsumers(module.getInputChannelName());
}
if (outputChannel != null) {
messageBus.unbindProducers(module.getOutputChannelName());
String tapChannelName = module.getTapChannelName();
messageBus.unbindProducers(tapChannelName);
}
}
protected final void bindChannels() {
Map<String, Object> historyProperties = null;
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);
// }
if (trackHistory) {
track(outputChannel, historyProperties);
}
}
if (inputChannel != null) {
bindMessageConsumer(inputChannel, module.getInputChannelName(),
module.getConsumerProperties());
if (trackHistory && outputChannel==null) {
track(inputChannel, historyProperties);
}
}
}
/*
* Following methods copied from parent to support the bindChannels() method above
*/
private void bindMessageConsumer(MessageChannel inputChannel,
String inputChannelName, Properties consumerProperties) {
if (isChannelPubSub(inputChannelName)) {
messageBus.bindPubSubConsumer(inputChannelName, inputChannel,
consumerProperties);
}
else {
messageBus.bindConsumer(inputChannelName, inputChannel, consumerProperties);
}
}
private void bindMessageProducer(MessageChannel outputChannel,
String outputChannelName, Properties producerProperties) {
if (isChannelPubSub(outputChannelName)) {
messageBus.bindPubSubProducer(outputChannelName, outputChannel,
producerProperties);
}
else {
messageBus.bindProducer(outputChannelName, outputChannel, producerProperties);
}
}
private boolean isChannelPubSub(String channelName) {
Assert.isTrue(StringUtils.hasText(channelName),
"Channel name should not be empty/null.");
return (channelName.startsWith("tap:") || channelName.startsWith("topic:"));
}
/**
* Creates a wiretap on the output channel and binds the tap channel to
* {@link MessageBus}'s message target.
*
* @param tapChannelName the name of the tap channel
* @param outputChannel the channel to tap
*/
private void createAndBindTapChannel(String tapChannelName,
MessageChannel outputChannel) {
logger.info("creating and binding tap channel for {}", tapChannelName);
if (outputChannel instanceof ChannelInterceptorAware) {
DirectChannel tapChannel = new DirectChannel();
tapChannel.setBeanName(tapChannelName + ".tap.bridge");
messageBus.bindPubSubProducer(tapChannelName, tapChannel, null); // TODO tap
// producer
// props
tapOutputChannel(tapChannel, (ChannelInterceptorAware) outputChannel);
}
else {
if (logger.isDebugEnabled()) {
logger.debug("output channel is not interceptor aware. Tap will not be created.");
}
}
}
private MessageChannel tapOutputChannel(MessageChannel tapChannel,
ChannelInterceptorAware outputChannel) {
outputChannel.addInterceptor(new WireTap(tapChannel));
return tapChannel;
}
private void track(MessageChannel channel, final Map<String, Object> historyProps) {
if (channel instanceof ChannelInterceptorAware) {
((ChannelInterceptorAware) channel)
.addInterceptor(new ChannelInterceptorAdapter() {
@Override
public Message<?> preSend(Message<?> message,
MessageChannel channel) {
@SuppressWarnings("unchecked")
Collection<Map<String, Object>> history = (Collection<Map<String, Object>>) message
.getHeaders().get(XdHeaders.XD_HISTORY);
if (history == null) {
history = new ArrayList<Map<String, Object>>(1);
}
else {
history = new ArrayList<Map<String, Object>>(history);
}
Map<String, Object> map = new LinkedHashMap<String, Object>();
map.putAll(historyProps);
map.put("thread", Thread.currentThread().getName());
history.add(map);
Message<?> out = messageBuilderFactory.fromMessage(message)
.setHeader(XdHeaders.XD_HISTORY, history).build();
return out;
}
});
}
}
}

50
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* 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.runner.adapter;
import org.springframework.messaging.MessageChannel;
/**
* @author Dave Syer
*
*/
public class OutputChannelSpec extends InputChannelSpec {
private boolean tapped;
private String tapChannelName;
public OutputChannelSpec(String name, MessageChannel channel) {
super(name, channel);
}
public boolean isTapped() {
return tapped;
}
public void setTapped(boolean tapped) {
this.tapped = tapped;
}
public void setTapChannelName(String tapChannelName) {
this.tapChannelName = tapChannelName;
}
public String getTapChannelName() {
return tapChannelName==null ? "" : tapChannelName;
}
}

41
spring-bus-core/src/main/java/org/springframework/bus/runner/config/LifecycleConfiguration.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* 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.runner.config;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.bus.runner.adapter.MessageBusAdapter;
import org.springframework.context.annotation.Configuration;
/**
* @author Dave Syer
*
*/
@Configuration
public class LifecycleConfiguration implements CommandLineRunner {
@Autowired
private MessageBusAdapter adapter;
@Override
public void run(String... args) throws Exception {
if (!adapter.isRunning()) {
adapter.start();
}
}
}

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

@@ -0,0 +1,73 @@
/*
* 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.runner.config;
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;
import org.springframework.xd.dirt.integration.bus.MessageBusAwareRouterBeanPostProcessor;
/**
* @author Dave Syer
*
*/
@Configuration
@Import(PropertyPlaceholderAutoConfiguration.class)
@ImportResource("classpath*:/META-INF/spring-xd/bus/codec.xml")
@EnableConfigurationProperties(MessageBusProperties.class)
public class MessageBusAdapterConfiguration {
@Autowired(required = false)
@Qualifier("output")
private MessageChannel output;
@Autowired(required = false)
@Qualifier("input")
private MessageChannel input;
@Bean
public MessageBusAdapter messageBusAdapter(MessageBusProperties module,
MessageBus messageBus) {
MessageBusAdapter adapter = new MessageBusAdapter(module, messageBus);
adapter.setOutputChannel(output);
adapter.setInputChannel(input);
return adapter;
}
// Nested class to avoid instantiating all of the above early
@Configuration
protected static class MessageBusAwareRouterConfiguration {
@Bean
public MessageBusAwareRouterBeanPostProcessor messageBusAwareRouterBeanPostProcessor(
MessageBus messageBus) {
return new MessageBusAwareRouterBeanPostProcessor(messageBus,
new Properties());
}
}
}

175
spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java Normal file
View File

@@ -0,0 +1,175 @@
/*
* 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.runner.config;
import java.util.Properties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.util.Assert;
import org.springframework.xd.dirt.integration.bus.BusUtils;
/**
* @author Dave Syer
*
*/
@ConfigurationProperties("spring.bus")
public class MessageBusProperties {
private String name = "module";
private String group = "group";
private int index = 0;
private String outputChannelName;
private String inputChannelName;
private String type = "processor";
private Properties consumerProperties = new Properties();
private Properties producerProperties = new Properties();
private Tap tap;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getGroup() {
return group;
}
public void setGroup(String group) {
this.group = group;
}
public int getIndex() {
return index;
}
public void setIndex(int index) {
this.index = index;
}
public String getInputChannelName() {
if (isTap()) {
return String.format("%s.%s.%s", BusUtils.constructTapPrefix(tap.getGroup()), tap.getName(), tap.getIndex());
}
return (inputChannelName != null) ? inputChannelName : BusUtils
.constructPipeName(group, index>0 ? index - 1 : index);
}
public String getOutputChannelName() {
return (outputChannelName != null) ? outputChannelName : BusUtils
.constructPipeName(group, index);
}
public String getTapChannelName() {
Assert.isTrue(!type.equals("job"), "Job module type not supported.");
// for Stream return channel name with indexed elements
return String.format("%s.%s.%s", BusUtils.constructTapPrefix(group), name, index);
}
public void setOutputChannelName(String outputChannelName) {
this.outputChannelName = outputChannelName;
}
public void setInputChannelName(String inputChannelName) {
this.inputChannelName = inputChannelName;
}
public String getType() {
return type;
}
public void setType(String type) {
this.type = type;
}
public Properties getConsumerProperties() {
return consumerProperties;
}
public void setConsumerProperties(Properties consumerProperties) {
this.consumerProperties = consumerProperties;
}
public Properties getProducerProperties() {
return producerProperties;
}
public void setProducerProperties(Properties producerProperties) {
this.producerProperties = producerProperties;
}
public Tap getTap() {
return tap;
}
public void setTap(Tap tap) {
this.tap = tap;
}
private boolean isTap() {
if (tap!=null) {
Assert.state(tap.getName()!=null, "Tap name not provided");
Assert.state(!tap.getGroup().equals(group), "Tap group cannot be the same as module group");
}
return tap!=null;
}
public static class Tap {
private String group = "group";
private String name;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
private int index = 0;
public String getGroup() {
return group;
}
public void setGroup(String group) {
this.group = group;
}
public int getIndex() {
return index;
}
public void setIndex(int index) {
this.index = index;
}
}
}

50
spring-bus-core/src/main/java/org/springframework/bus/runner/config/RabbitServiceConfiguration.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* 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.runner.config;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.cloud.config.java.AbstractCloudConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
import org.springframework.context.annotation.Profile;
import org.springframework.xd.dirt.integration.rabbit.RabbitMessageBus;
/**
* Bind to services, either locally or in a Lattice environment.
*
* @author Mark Fisher
* @author Dave Syer
*/
@Configuration
@ConditionalOnClass(RabbitMessageBus.class)
@ImportResource({ "classpath*:/META-INF/spring-xd/bus/rabbit-bus.xml",
"classpath*:/META-INF/spring-xd/analytics/rabbit-analytics.xml" })
public class RabbitServiceConfiguration {
@Configuration
@Profile("cloud")
protected static class CloudConfig extends AbstractCloudConfig {
@Bean
ConnectionFactory rabbitConnectionFactory() {
return connectionFactory().rabbitConnectionFactory();
}
}
}

50
spring-bus-core/src/main/java/org/springframework/bus/runner/config/RedisServiceConfiguration.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* 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.runner.config;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.cloud.config.java.AbstractCloudConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportResource;
import org.springframework.context.annotation.Profile;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.xd.dirt.integration.redis.RedisMessageBus;
/**
* Bind to services, either locally or in a Lattice environment.
*
* @author Mark Fisher
* @author Dave Syer
*/
@Configuration
@ConditionalOnClass(RedisMessageBus.class)
@ImportResource({ "classpath*:/META-INF/spring-xd/bus/redis-bus.xml",
"classpath*:/META-INF/spring-xd/analytics/redis-analytics.xml" })
public class RedisServiceConfiguration {
@Configuration
@Profile("cloud")
protected static class CloudConfig extends AbstractCloudConfig {
@Bean
RedisConnectionFactory redisConnectionFactory() {
return connectionFactory().redisConnectionFactory();
}
}
}

46
spring-xd-runner/pom.xml Normal file
View File

@@ -0,0 +1,46 @@
<?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.bus</groupId>
<artifactId>spring-xd-runner</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<name>spring-xd-module-runner</name>
<description>Demo project for Spring XD Modules as apps</description>
<parent>
<groupId>org.springframework.bus</groupId>
<artifactId>spring-bus-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-context</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.bus</groupId>
<artifactId>spring-bus-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-dirt</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

2
spring-xd-runner/src/main/resources/META-INF/spring.factories Normal file
View File

@@ -0,0 +1,2 @@
org.springframework.cloud.bootstrap.BootstrapConfiguration:\
org.springframework.bus.xd.bootstrap.ModuleOptionsPropertySourceInitializer

24
spring-xd-samples/pom.xml Normal file
View File

@@ -0,0 +1,24 @@
<?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.bus</groupId>
<artifactId>spring-xd-samples</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>pom</packaging>
<url>http://projects.spring.io/spring-xd/</url>
<organization>
<name>Pivotal Software, Inc.</name>
<url>http://www.spring.io</url>
</organization>
<parent>
<groupId>org.springframework.bus</groupId>
<artifactId>spring-bus-parent</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<modules>
<module>source</module>
<module>sink</module>
<module>tap</module>
<module>source-xml</module>
</modules>
</project>

57
spring-xd-samples/sink/pom.xml Normal file
View File

@@ -0,0 +1,57 @@
<?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.bus</groupId>
<artifactId>spring-xd-module-runner-sample-sink</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<name>spring-xd-module-runner-sample-sink</name>
<description>Demo project for Spring XD module</description>
<parent>
<groupId>org.springframework.bus</groupId>
<artifactId>spring-bus-parent</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.bus</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-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>

48
spring-xd-samples/sink/src/main/java/config/ModuleDefinition.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* 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 config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;
/**
* @author Dave Syer
*
*/
@Configuration
@MessageEndpoint
public class ModuleDefinition {
private static Logger logger = LoggerFactory.getLogger(ModuleDefinition.class);
@Bean
public MessageChannel input() {
return new DirectChannel();
}
@ServiceActivator(inputChannel="input")
public void loggerSink(Object payload) {
logger.info("Received: " + payload);
}
}

19
spring-xd-samples/sink/src/main/java/demo/SinkApplication.java Normal file
View File

@@ -0,0 +1,19 @@
package demo;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.bus.runner.EnableMessageBus;
import org.springframework.context.annotation.ComponentScan;
import config.ModuleDefinition;
@SpringBootApplication
@EnableMessageBus
@ComponentScan(basePackageClasses=ModuleDefinition.class)
public class SinkApplication {
public static void main(String[] args) throws InterruptedException {
SpringApplication.run(SinkApplication.class, args);
}
}

16
spring-xd-samples/sink/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,16 @@
server:
port: 8081
---
xd:
messagebus:
redis:
headers:
# comma-delimited list of additional (string-valued) header names to transport
default:
# default bus properties, if not specified at the module level
backOffInitialInterval: 1000
backOffMaxInterval: 10000
backOffMultiplier: 2.0
concurrency: 1
maxAttempts: 3

6
spring-xd-samples/sink/src/main/resources/bootstrap.yml Normal file
View File

@@ -0,0 +1,6 @@
---
spring:
bus:
group: testtock
name: logger
index: 1

1
spring-xd-samples/sink/src/main/resources/config/logger.properties Normal file
View File

@@ -0,0 +1 @@
base_packages = config

20
spring-xd-samples/sink/src/test/java/demo/ModuleApplicationTests.java Normal file
View File

@@ -0,0 +1,20 @@
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 = SinkApplication.class)
@WebAppConfiguration
@DirtiesContext
public class ModuleApplicationTests {
@Test
public void contextLoads() {
}
}

56
spring-xd-samples/source-xml/pom.xml Normal file
View File

@@ -0,0 +1,56 @@
<?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.bus</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.bus</groupId>
<artifactId>spring-bus-parent</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.bus</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-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 Normal file
View File

@@ -0,0 +1,19 @@
package demo;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.bus.runner.EnableMessageBus;
import org.springframework.context.annotation.ImportResource;
import org.springframework.context.annotation.PropertySource;
@SpringBootApplication
@EnableMessageBus
@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);
}
}

15
spring-xd-samples/source-xml/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,15 @@
fixedDelay: 5
---
xd:
messagebus:
redis:
headers:
# comma-delimited list of additional (string-valued) header names to transport
default:
# default bus properties, if not specified at the module level
backOffInitialInterval: 1000
backOffMaxInterval: 10000
backOffMultiplier: 2.0
concurrency: 1
maxAttempts: 3

5
spring-xd-samples/source-xml/src/main/resources/bootstrap.yml Normal file
View File

@@ -0,0 +1,5 @@
---
spring:
bus:
group: testtock
name: ticker

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

@@ -0,0 +1 @@
options_class = org.springframework.xd.dirt.modules.metadata.TimeSourceOptionsMetadata

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

@@ -0,0 +1,22 @@
<?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 Normal file
View File

@@ -0,0 +1,20 @@
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() {
}
}

56
spring-xd-samples/source/pom.xml Normal file
View File

@@ -0,0 +1,56 @@
<?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.bus</groupId>
<artifactId>spring-xd-module-runner-sample-source</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<name>spring-xd-module-runner-sample-source</name>
<description>Demo project for Spring XD module</description>
<parent>
<groupId>org.springframework.bus</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.bus</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-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>

53
spring-xd-samples/source/src/main/java/config/ModuleDefinition.java Normal file
View File

@@ -0,0 +1,53 @@
/*
* 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 config;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Dave Syer
*
*/
@Configuration
public class ModuleDefinition {
@Value("${format}")
private String format;
@Bean
public MessageChannel output() {
return new DirectChannel();
}
@Bean
@InboundChannelAdapter(value = "output", autoStartup = "false", poller = @Poller(fixedDelay = "${fixedDelay}", maxMessagesPerPoll = "1"))
public MessageSource<String> timerMessageSource() {
return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date()));
}
}

19
spring-xd-samples/source/src/main/java/demo/SourceApplication.java Normal file
View File

@@ -0,0 +1,19 @@
package demo;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.bus.runner.EnableMessageBus;
import org.springframework.context.annotation.ComponentScan;
import config.ModuleDefinition;
@SpringBootApplication
@EnableMessageBus
@ComponentScan(basePackageClasses=ModuleDefinition.class)
public class SourceApplication {
public static void main(String[] args) throws InterruptedException {
SpringApplication.run(SourceApplication.class, args);
}
}

15
spring-xd-samples/source/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,15 @@
fixedDelay: 5000
---
xd:
messagebus:
redis:
headers:
# comma-delimited list of additional (string-valued) header names to transport
default:
# default bus properties, if not specified at the module level
backOffInitialInterval: 1000
backOffMaxInterval: 10000
backOffMultiplier: 2.0
concurrency: 1
maxAttempts: 3

5
spring-xd-samples/source/src/main/resources/bootstrap.yml Normal file
View File

@@ -0,0 +1,5 @@
---
spring:
bus:
group: testtock
name: ticker

2
spring-xd-samples/source/src/main/resources/config/ticker.properties Normal file
View File

@@ -0,0 +1,2 @@
options_class = org.springframework.xd.dirt.modules.metadata.TimeSourceOptionsMetadata
base_packages = config

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

@@ -0,0 +1,20 @@
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 = SourceApplication.class)
@WebAppConfiguration
@DirtiesContext
public class ModuleApplicationTests {
@Test
public void contextLoads() {
}
}

56
spring-xd-samples/tap/pom.xml Normal file
View File

@@ -0,0 +1,56 @@
<?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.xd</groupId>
<artifactId>spring-xd-module-runner-sample-tap</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<packaging>jar</packaging>
<name>spring-xd-module-runner-sample-tap</name>
<description>Demo project for Spring XD module</description>
<parent>
<groupId>org.springframework.bus</groupId>
<artifactId>spring-bus-parent</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.bus</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-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>

48
spring-xd-samples/tap/src/main/java/config/ModuleDefinition.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* 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 config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;
/**
* @author Dave Syer
*
*/
@Configuration
@MessageEndpoint
public class ModuleDefinition {
private static Logger logger = LoggerFactory.getLogger(ModuleDefinition.class);
@Bean
public MessageChannel input() {
return new DirectChannel();
}
@ServiceActivator(inputChannel="input")
public void loggerSink(Object payload) {
logger.info("Received: " + payload);
}
}

19
spring-xd-samples/tap/src/main/java/demo/TapApplication.java Normal file
View File

@@ -0,0 +1,19 @@
package demo;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.bus.runner.EnableMessageBus;
import org.springframework.context.annotation.ComponentScan;
import config.ModuleDefinition;
@SpringBootApplication
@EnableMessageBus
@ComponentScan(basePackageClasses=ModuleDefinition.class)
public class TapApplication {
public static void main(String[] args) throws InterruptedException {
SpringApplication.run(TapApplication.class, args);
}
}

16
spring-xd-samples/tap/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,16 @@
server:
port: 8081
---
xd:
messagebus:
redis:
headers:
# comma-delimited list of additional (string-valued) header names to transport
default:
# default bus properties, if not specified at the module level
backOffInitialInterval: 1000
backOffMaxInterval: 10000
backOffMultiplier: 2.0
concurrency: 1
maxAttempts: 3

11
spring-xd-samples/tap/src/main/resources/bootstrap.yml Normal file
View File

@@ -0,0 +1,11 @@
---
spring:
bus:
group: tocktap
name: logger
index: 0
tap:
group: testtock
name: ticker
index: 0

1
spring-xd-samples/tap/src/main/resources/config/logger.properties Normal file
View File

@@ -0,0 +1 @@
base_packages = config

20
spring-xd-samples/tap/src/test/java/demo/ModuleApplicationTests.java Normal file
View File

@@ -0,0 +1,20 @@
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 = TapApplication.class)
@WebAppConfiguration
@DirtiesContext
public class ModuleApplicationTests {
@Test
public void contextLoads() {
}
}