From f58703acdb260c10952b0a63e1064861ed0f1ff0 Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Thu, 28 May 2015 16:42:00 +0100 Subject: [PATCH] Migrate some code from prototype --- .gitignore | 20 ++ README.md | 105 ++++++- pom.xml | 142 ++++++++++ roadmap.md | 114 ++++++++ spring-bus-core/pom.xml | 62 ++++ .../bus/runner/EnableMessageBus.java | 46 +++ .../bus/runner/adapter/InputChannelSpec.java | 43 +++ .../bus/runner/adapter/MessageBusAdapter.java | 267 ++++++++++++++++++ .../bus/runner/adapter/OutputChannelSpec.java | 50 ++++ .../runner/config/LifecycleConfiguration.java | 41 +++ .../MessageBusAdapterConfiguration.java | 73 +++++ .../runner/config/MessageBusProperties.java | 175 ++++++++++++ .../config/RabbitServiceConfiguration.java | 50 ++++ .../config/RedisServiceConfiguration.java | 50 ++++ spring-xd-runner/pom.xml | 46 +++ .../main/resources/META-INF/spring.factories | 2 + spring-xd-samples/pom.xml | 24 ++ spring-xd-samples/sink/pom.xml | 57 ++++ .../main/java/config/ModuleDefinition.java | 48 ++++ .../src/main/java/demo/SinkApplication.java | 19 ++ .../sink/src/main/resources/application.yml | 16 ++ .../sink/src/main/resources/bootstrap.yml | 6 + .../main/resources/config/logger.properties | 1 + .../java/demo/ModuleApplicationTests.java | 20 ++ spring-xd-samples/source-xml/pom.xml | 56 ++++ .../src/main/java/demo/ModuleApplication.java | 19 ++ .../src/main/resources/application.yml | 15 + .../src/main/resources/bootstrap.yml | 5 + .../main/resources/config/ticker.properties | 1 + .../src/main/resources/config/ticker.xml | 22 ++ .../java/demo/ModuleApplicationTests.java | 20 ++ spring-xd-samples/source/pom.xml | 56 ++++ .../main/java/config/ModuleDefinition.java | 53 ++++ .../src/main/java/demo/SourceApplication.java | 19 ++ .../source/src/main/resources/application.yml | 15 + .../source/src/main/resources/bootstrap.yml | 5 + .../main/resources/config/ticker.properties | 2 + .../java/demo/ModuleApplicationTests.java | 20 ++ spring-xd-samples/tap/pom.xml | 56 ++++ .../main/java/config/ModuleDefinition.java | 48 ++++ .../src/main/java/demo/TapApplication.java | 19 ++ .../tap/src/main/resources/application.yml | 16 ++ .../tap/src/main/resources/bootstrap.yml | 11 + .../main/resources/config/logger.properties | 1 + .../java/demo/ModuleApplicationTests.java | 20 ++ 45 files changed, 1955 insertions(+), 1 deletion(-) create mode 100644 .gitignore create mode 100644 pom.xml create mode 100644 roadmap.md create mode 100644 spring-bus-core/pom.xml create mode 100644 spring-bus-core/src/main/java/org/springframework/bus/runner/EnableMessageBus.java create mode 100644 spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java create mode 100644 spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java create mode 100644 spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java create mode 100644 spring-bus-core/src/main/java/org/springframework/bus/runner/config/LifecycleConfiguration.java create mode 100644 spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java create mode 100644 spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java create mode 100644 spring-bus-core/src/main/java/org/springframework/bus/runner/config/RabbitServiceConfiguration.java create mode 100644 spring-bus-core/src/main/java/org/springframework/bus/runner/config/RedisServiceConfiguration.java create mode 100644 spring-xd-runner/pom.xml create mode 100644 spring-xd-runner/src/main/resources/META-INF/spring.factories create mode 100644 spring-xd-samples/pom.xml create mode 100644 spring-xd-samples/sink/pom.xml create mode 100644 spring-xd-samples/sink/src/main/java/config/ModuleDefinition.java create mode 100644 spring-xd-samples/sink/src/main/java/demo/SinkApplication.java create mode 100644 spring-xd-samples/sink/src/main/resources/application.yml create mode 100644 spring-xd-samples/sink/src/main/resources/bootstrap.yml create mode 100644 spring-xd-samples/sink/src/main/resources/config/logger.properties create mode 100644 spring-xd-samples/sink/src/test/java/demo/ModuleApplicationTests.java create mode 100644 spring-xd-samples/source-xml/pom.xml create mode 100644 spring-xd-samples/source-xml/src/main/java/demo/ModuleApplication.java create mode 100644 spring-xd-samples/source-xml/src/main/resources/application.yml create mode 100644 spring-xd-samples/source-xml/src/main/resources/bootstrap.yml create mode 100644 spring-xd-samples/source-xml/src/main/resources/config/ticker.properties create mode 100644 spring-xd-samples/source-xml/src/main/resources/config/ticker.xml create mode 100644 spring-xd-samples/source-xml/src/test/java/demo/ModuleApplicationTests.java create mode 100644 spring-xd-samples/source/pom.xml create mode 100644 spring-xd-samples/source/src/main/java/config/ModuleDefinition.java create mode 100644 spring-xd-samples/source/src/main/java/demo/SourceApplication.java create mode 100644 spring-xd-samples/source/src/main/resources/application.yml create mode 100644 spring-xd-samples/source/src/main/resources/bootstrap.yml create mode 100644 spring-xd-samples/source/src/main/resources/config/ticker.properties create mode 100644 spring-xd-samples/source/src/test/java/demo/ModuleApplicationTests.java create mode 100644 spring-xd-samples/tap/pom.xml create mode 100644 spring-xd-samples/tap/src/main/java/config/ModuleDefinition.java create mode 100644 spring-xd-samples/tap/src/main/java/demo/TapApplication.java create mode 100644 spring-xd-samples/tap/src/main/resources/application.yml create mode 100644 spring-xd-samples/tap/src/main/resources/bootstrap.yml create mode 100644 spring-xd-samples/tap/src/main/resources/config/logger.properties create mode 100644 spring-xd-samples/tap/src/test/java/demo/ModuleApplicationTests.java diff --git a/.gitignore b/.gitignore new file mode 100644 index 000000000..41884cf75 --- /dev/null +++ b/.gitignore @@ -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 diff --git a/README.md b/README.md index 4bd7f0895..ca9794c26 100644 --- a/README.md +++ b/README.md @@ -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 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..` 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:..` 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. \ No newline at end of file diff --git a/pom.xml b/pom.xml new file mode 100644 index 000000000..6fa73298c --- /dev/null +++ b/pom.xml @@ -0,0 +1,142 @@ + + + 4.0.0 + org.springframework.bus + spring-bus-parent + 1.0.0.BUILD-SNAPSHOT + pom + http://projects.spring.io/spring-xd/ + + Pivotal Software, Inc. + http://www.spring.io + + + org.springframework.boot + spring-boot-starter-parent + 1.2.3.RELEASE + + + + spring-bus-core + spring-xd-runner + spring-xd-samples + + + + + org.springframework.cloud + spring-cloud-starter-parent + 1.0.2.RELEASE + pom + import + + + org.springframework.bus + spring-bus-core + 1.0.0.BUILD-SNAPSHOT + + + org.springframework.bus + spring-xd-runner + 1.0.0.BUILD-SNAPSHOT + + + org.springframework.xd + spring-xd-dirt + 1.2.0.BUILD-SNAPSHOT + + + org.springframework.xd + spring-xd-spark-streaming + + + org.springframework.xd + spring-xd-hadoop + + + org.springframework.xd + spring-xd-batch + + + org.springframework.xd + spring-xd-ui + + + slf4j-log4j12 + org.slf4j + + + zookeeper + org.apache.zookeeper + + + spring-security-ldap + org.springframework.security + + + spring-jdbc + org.springframework + + + spring-batch-integration + org.springframework.batch + + + spring-batch-admin-manager + org.springframework.batch + + + spring-data-mongodb + org.springframework.data + + + + + org.springframework.xd + spring-xd-messagebus-redis + 1.2.0.BUILD-SNAPSHOT + + + org.springframework.xd + spring-xd-messagebus-rabbit + 1.2.0.BUILD-SNAPSHOT + + + + + + spring-snapshots + Spring Snapshots + http://repo.spring.io/snapshot + + true + + + + spring-milestones + Spring Milestones + http://repo.spring.io/milestone + + false + + + + + + spring-snapshots + Spring Snapshots + http://repo.spring.io/snapshot + + true + + + + spring-milestones + Spring Milestones + http://repo.spring.io/milestone + + false + + + + diff --git a/roadmap.md b/roadmap.md new file mode 100644 index 000000000..192002590 --- /dev/null +++ b/roadmap.md @@ -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 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). diff --git a/spring-bus-core/pom.xml b/spring-bus-core/pom.xml new file mode 100644 index 000000000..2d6071a6b --- /dev/null +++ b/spring-bus-core/pom.xml @@ -0,0 +1,62 @@ + + + 4.0.0 + + org.springframework.bus + spring-bus-core + 1.0.0.BUILD-SNAPSHOT + jar + + spring-bus-runner + Demo project for Spring Integration as apps + + + org.springframework.bus + spring-bus-parent + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + 1.8 + + + + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.xd + spring-xd-dirt + + + + org.springframework.xd + spring-xd-messagebus-redis + true + + + org.springframework.xd + spring-xd-messagebus-rabbit + true + + + org.springframework.boot + spring-boot-configuration-processor + true + + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/EnableMessageBus.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/EnableMessageBus.java new file mode 100644 index 000000000..ffdbc3137 --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/EnableMessageBus.java @@ -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 { + +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java new file mode 100644 index 000000000..888da6769 --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java @@ -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; + } + +} 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 new file mode 100644 index 000000000..48f7ec525 --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java @@ -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 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 historyProps) { + if (channel instanceof ChannelInterceptorAware) { + ((ChannelInterceptorAware) channel) + .addInterceptor(new ChannelInterceptorAdapter() { + + @Override + public Message preSend(Message message, + MessageChannel channel) { + @SuppressWarnings("unchecked") + Collection> history = (Collection>) message + .getHeaders().get(XdHeaders.XD_HISTORY); + if (history == null) { + history = new ArrayList>(1); + } + else { + history = new ArrayList>(history); + } + Map map = new LinkedHashMap(); + 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; + } + }); + } + } +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java new file mode 100644 index 000000000..0fe0c2481 --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java @@ -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; + } + +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/LifecycleConfiguration.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/LifecycleConfiguration.java new file mode 100644 index 000000000..b97b8734f --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/LifecycleConfiguration.java @@ -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(); + } + } + +} 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 new file mode 100644 index 000000000..845e03b27 --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java @@ -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()); + } + + } + +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java new file mode 100644 index 000000000..400a614fa --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java @@ -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; + } + + } + +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/RabbitServiceConfiguration.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/RabbitServiceConfiguration.java new file mode 100644 index 000000000..3146acd67 --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/RabbitServiceConfiguration.java @@ -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(); + } + } + +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/RedisServiceConfiguration.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/RedisServiceConfiguration.java new file mode 100644 index 000000000..28328f7fe --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/RedisServiceConfiguration.java @@ -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(); + } + } + +} diff --git a/spring-xd-runner/pom.xml b/spring-xd-runner/pom.xml new file mode 100644 index 000000000..cc26c6953 --- /dev/null +++ b/spring-xd-runner/pom.xml @@ -0,0 +1,46 @@ + + + 4.0.0 + + org.springframework.bus + spring-xd-runner + 1.0.0.BUILD-SNAPSHOT + jar + + spring-xd-module-runner + Demo project for Spring XD Modules as apps + + + org.springframework.bus + spring-bus-parent + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + 1.8 + + + + + org.springframework.cloud + spring-cloud-context + + + org.springframework.bus + spring-bus-core + + + org.springframework.xd + spring-xd-dirt + + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/spring-xd-runner/src/main/resources/META-INF/spring.factories b/spring-xd-runner/src/main/resources/META-INF/spring.factories new file mode 100644 index 000000000..b4c8230a6 --- /dev/null +++ b/spring-xd-runner/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.cloud.bootstrap.BootstrapConfiguration:\ +org.springframework.bus.xd.bootstrap.ModuleOptionsPropertySourceInitializer \ No newline at end of file diff --git a/spring-xd-samples/pom.xml b/spring-xd-samples/pom.xml new file mode 100644 index 000000000..5bfd7c1ef --- /dev/null +++ b/spring-xd-samples/pom.xml @@ -0,0 +1,24 @@ + + + 4.0.0 + org.springframework.bus + spring-xd-samples + 1.0.0.BUILD-SNAPSHOT + pom + http://projects.spring.io/spring-xd/ + + Pivotal Software, Inc. + http://www.spring.io + + + org.springframework.bus + spring-bus-parent + 1.0.0.BUILD-SNAPSHOT + + + source + sink + tap + source-xml + + diff --git a/spring-xd-samples/sink/pom.xml b/spring-xd-samples/sink/pom.xml new file mode 100644 index 000000000..4557091db --- /dev/null +++ b/spring-xd-samples/sink/pom.xml @@ -0,0 +1,57 @@ + + + 4.0.0 + + org.springframework.bus + spring-xd-module-runner-sample-sink + 1.0.0.BUILD-SNAPSHOT + jar + + spring-xd-module-runner-sample-sink + Demo project for Spring XD module + + + org.springframework.bus + spring-bus-parent + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + demo.ModuleApplication + 1.8 + + + + + org.springframework.bus + spring-xd-runner + + + org.springframework.xd + spring-xd-messagebus-redis + + + org.springframework.boot + spring-boot-configuration-processor + true + + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + diff --git a/spring-xd-samples/sink/src/main/java/config/ModuleDefinition.java b/spring-xd-samples/sink/src/main/java/config/ModuleDefinition.java new file mode 100644 index 000000000..245cc76f5 --- /dev/null +++ b/spring-xd-samples/sink/src/main/java/config/ModuleDefinition.java @@ -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); + } + +} diff --git a/spring-xd-samples/sink/src/main/java/demo/SinkApplication.java b/spring-xd-samples/sink/src/main/java/demo/SinkApplication.java new file mode 100644 index 000000000..c15d582e7 --- /dev/null +++ b/spring-xd-samples/sink/src/main/java/demo/SinkApplication.java @@ -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); + } + +} diff --git a/spring-xd-samples/sink/src/main/resources/application.yml b/spring-xd-samples/sink/src/main/resources/application.yml new file mode 100644 index 000000000..17fd1f724 --- /dev/null +++ b/spring-xd-samples/sink/src/main/resources/application.yml @@ -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 diff --git a/spring-xd-samples/sink/src/main/resources/bootstrap.yml b/spring-xd-samples/sink/src/main/resources/bootstrap.yml new file mode 100644 index 000000000..25fe1436c --- /dev/null +++ b/spring-xd-samples/sink/src/main/resources/bootstrap.yml @@ -0,0 +1,6 @@ +--- +spring: + bus: + group: testtock + name: logger + index: 1 diff --git a/spring-xd-samples/sink/src/main/resources/config/logger.properties b/spring-xd-samples/sink/src/main/resources/config/logger.properties new file mode 100644 index 000000000..b13888b25 --- /dev/null +++ b/spring-xd-samples/sink/src/main/resources/config/logger.properties @@ -0,0 +1 @@ +base_packages = config \ No newline at end of file diff --git a/spring-xd-samples/sink/src/test/java/demo/ModuleApplicationTests.java b/spring-xd-samples/sink/src/test/java/demo/ModuleApplicationTests.java new file mode 100644 index 000000000..f2d65e89f --- /dev/null +++ b/spring-xd-samples/sink/src/test/java/demo/ModuleApplicationTests.java @@ -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() { + } + +} diff --git a/spring-xd-samples/source-xml/pom.xml b/spring-xd-samples/source-xml/pom.xml new file mode 100644 index 000000000..34ccb6757 --- /dev/null +++ b/spring-xd-samples/source-xml/pom.xml @@ -0,0 +1,56 @@ + + + 4.0.0 + + org.springframework.bus + spring-xd-module-runner-sample-source-xml + 1.0.0.BUILD-SNAPSHOT + jar + + spring-xd-module-runner-sample-source-xml + Demo project for Spring XD module + + + org.springframework.bus + spring-bus-parent + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + demo.ModuleApplication + 1.8 + + + + + org.springframework.bus + spring-xd-runner + + + org.springframework.xd + spring-xd-messagebus-redis + + + org.springframework.boot + spring-boot-configuration-processor + true + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + diff --git a/spring-xd-samples/source-xml/src/main/java/demo/ModuleApplication.java b/spring-xd-samples/source-xml/src/main/java/demo/ModuleApplication.java new file mode 100644 index 000000000..03007d978 --- /dev/null +++ b/spring-xd-samples/source-xml/src/main/java/demo/ModuleApplication.java @@ -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); + } + +} diff --git a/spring-xd-samples/source-xml/src/main/resources/application.yml b/spring-xd-samples/source-xml/src/main/resources/application.yml new file mode 100644 index 000000000..a697962f0 --- /dev/null +++ b/spring-xd-samples/source-xml/src/main/resources/application.yml @@ -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 diff --git a/spring-xd-samples/source-xml/src/main/resources/bootstrap.yml b/spring-xd-samples/source-xml/src/main/resources/bootstrap.yml new file mode 100644 index 000000000..0acc7746d --- /dev/null +++ b/spring-xd-samples/source-xml/src/main/resources/bootstrap.yml @@ -0,0 +1,5 @@ +--- +spring: + bus: + group: testtock + name: ticker diff --git a/spring-xd-samples/source-xml/src/main/resources/config/ticker.properties b/spring-xd-samples/source-xml/src/main/resources/config/ticker.properties new file mode 100644 index 000000000..96c0017ad --- /dev/null +++ b/spring-xd-samples/source-xml/src/main/resources/config/ticker.properties @@ -0,0 +1 @@ +options_class = org.springframework.xd.dirt.modules.metadata.TimeSourceOptionsMetadata \ No newline at end of file diff --git a/spring-xd-samples/source-xml/src/main/resources/config/ticker.xml b/spring-xd-samples/source-xml/src/main/resources/config/ticker.xml new file mode 100644 index 000000000..7b59ada64 --- /dev/null +++ b/spring-xd-samples/source-xml/src/main/resources/config/ticker.xml @@ -0,0 +1,22 @@ + + + + + + + + + + + + + + + + diff --git a/spring-xd-samples/source-xml/src/test/java/demo/ModuleApplicationTests.java b/spring-xd-samples/source-xml/src/test/java/demo/ModuleApplicationTests.java new file mode 100644 index 000000000..60627489e --- /dev/null +++ b/spring-xd-samples/source-xml/src/test/java/demo/ModuleApplicationTests.java @@ -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() { + } + +} diff --git a/spring-xd-samples/source/pom.xml b/spring-xd-samples/source/pom.xml new file mode 100644 index 000000000..5613388a9 --- /dev/null +++ b/spring-xd-samples/source/pom.xml @@ -0,0 +1,56 @@ + + + 4.0.0 + + org.springframework.bus + spring-xd-module-runner-sample-source + 1.0.0.BUILD-SNAPSHOT + jar + + spring-xd-module-runner-sample-source + Demo project for Spring XD module + + + org.springframework.bus + spring-xd-samples + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + demo.ModuleApplication + 1.8 + + + + + org.springframework.bus + spring-xd-runner + + + org.springframework.xd + spring-xd-messagebus-redis + + + org.springframework.boot + spring-boot-configuration-processor + true + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + diff --git a/spring-xd-samples/source/src/main/java/config/ModuleDefinition.java b/spring-xd-samples/source/src/main/java/config/ModuleDefinition.java new file mode 100644 index 000000000..9d4e93c62 --- /dev/null +++ b/spring-xd-samples/source/src/main/java/config/ModuleDefinition.java @@ -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 timerMessageSource() { + return () -> new GenericMessage<>(new SimpleDateFormat(format).format(new Date())); + } + +} diff --git a/spring-xd-samples/source/src/main/java/demo/SourceApplication.java b/spring-xd-samples/source/src/main/java/demo/SourceApplication.java new file mode 100644 index 000000000..f4011f4bd --- /dev/null +++ b/spring-xd-samples/source/src/main/java/demo/SourceApplication.java @@ -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); + } + +} diff --git a/spring-xd-samples/source/src/main/resources/application.yml b/spring-xd-samples/source/src/main/resources/application.yml new file mode 100644 index 000000000..1d4d2eefe --- /dev/null +++ b/spring-xd-samples/source/src/main/resources/application.yml @@ -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 diff --git a/spring-xd-samples/source/src/main/resources/bootstrap.yml b/spring-xd-samples/source/src/main/resources/bootstrap.yml new file mode 100644 index 000000000..0acc7746d --- /dev/null +++ b/spring-xd-samples/source/src/main/resources/bootstrap.yml @@ -0,0 +1,5 @@ +--- +spring: + bus: + group: testtock + name: ticker diff --git a/spring-xd-samples/source/src/main/resources/config/ticker.properties b/spring-xd-samples/source/src/main/resources/config/ticker.properties new file mode 100644 index 000000000..464d6dcc4 --- /dev/null +++ b/spring-xd-samples/source/src/main/resources/config/ticker.properties @@ -0,0 +1,2 @@ +options_class = org.springframework.xd.dirt.modules.metadata.TimeSourceOptionsMetadata +base_packages = config \ No newline at end of file diff --git a/spring-xd-samples/source/src/test/java/demo/ModuleApplicationTests.java b/spring-xd-samples/source/src/test/java/demo/ModuleApplicationTests.java new file mode 100644 index 000000000..23a9cf3b0 --- /dev/null +++ b/spring-xd-samples/source/src/test/java/demo/ModuleApplicationTests.java @@ -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() { + } + +} diff --git a/spring-xd-samples/tap/pom.xml b/spring-xd-samples/tap/pom.xml new file mode 100644 index 000000000..38bf36f8f --- /dev/null +++ b/spring-xd-samples/tap/pom.xml @@ -0,0 +1,56 @@ + + + 4.0.0 + + org.springframework.xd + spring-xd-module-runner-sample-tap + 1.0.0.BUILD-SNAPSHOT + jar + + spring-xd-module-runner-sample-tap + Demo project for Spring XD module + + + org.springframework.bus + spring-bus-parent + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + demo.ModuleApplication + 1.8 + + + + + org.springframework.bus + spring-xd-runner + + + org.springframework.xd + spring-xd-messagebus-redis + + + org.springframework.boot + spring-boot-configuration-processor + true + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + diff --git a/spring-xd-samples/tap/src/main/java/config/ModuleDefinition.java b/spring-xd-samples/tap/src/main/java/config/ModuleDefinition.java new file mode 100644 index 000000000..245cc76f5 --- /dev/null +++ b/spring-xd-samples/tap/src/main/java/config/ModuleDefinition.java @@ -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); + } + +} diff --git a/spring-xd-samples/tap/src/main/java/demo/TapApplication.java b/spring-xd-samples/tap/src/main/java/demo/TapApplication.java new file mode 100644 index 000000000..1c6f957db --- /dev/null +++ b/spring-xd-samples/tap/src/main/java/demo/TapApplication.java @@ -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); + } + +} diff --git a/spring-xd-samples/tap/src/main/resources/application.yml b/spring-xd-samples/tap/src/main/resources/application.yml new file mode 100644 index 000000000..17fd1f724 --- /dev/null +++ b/spring-xd-samples/tap/src/main/resources/application.yml @@ -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 diff --git a/spring-xd-samples/tap/src/main/resources/bootstrap.yml b/spring-xd-samples/tap/src/main/resources/bootstrap.yml new file mode 100644 index 000000000..698dade5b --- /dev/null +++ b/spring-xd-samples/tap/src/main/resources/bootstrap.yml @@ -0,0 +1,11 @@ +--- +spring: + bus: + group: tocktap + name: logger + index: 0 + tap: + group: testtock + name: ticker + index: 0 + \ No newline at end of file diff --git a/spring-xd-samples/tap/src/main/resources/config/logger.properties b/spring-xd-samples/tap/src/main/resources/config/logger.properties new file mode 100644 index 000000000..b13888b25 --- /dev/null +++ b/spring-xd-samples/tap/src/main/resources/config/logger.properties @@ -0,0 +1 @@ +base_packages = config \ No newline at end of file diff --git a/spring-xd-samples/tap/src/test/java/demo/ModuleApplicationTests.java b/spring-xd-samples/tap/src/test/java/demo/ModuleApplicationTests.java new file mode 100644 index 000000000..e8c471407 --- /dev/null +++ b/spring-xd-samples/tap/src/test/java/demo/ModuleApplicationTests.java @@ -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() { + } + +}