Support RxJava module

- Add a way to enable `RxJava` module
   - Add `EnableRxJavaModule` annotation that inherits `EnableModule` for `Processor` type.
      - It also registers the custom `MessageHandler` that sets up RxJava `Observable/Subscriber` to process the incoming message reactively.
      - Introduce `RxJavaProcessor` that is expected to be implemented by the module author with the input/output being `Observable`. This processor is autowired into the message handler registered above.
 - Add moving average sample

Add project `spring-cloud-stream-rxjava`

 - This project extends `spring-cloud-stream` so that any RxJava processing module would have this as a dependency

pom.xml

@@ -19,6 +19,7 @@
 	</scm>
 	<properties>
 		<java.version>1.7</java.version>
+		<rx-java.version>1.0.14</rx-java.version>
 		<spring-boot.version>1.3.0.BUILD-SNAPSHOT</spring-boot.version>
 		<spring-framework.version>4.2.1.BUILD-SNAPSHOT</spring-framework.version>
 		<spring-integration.version>4.2.0.BUILD-SNAPSHOT</spring-integration.version>
@@ -28,6 +29,7 @@
 		<module>spring-cloud-stream-binders</module>
 		<module>spring-cloud-stream-codec</module>
 		<module>spring-cloud-stream-starters</module>
+		<module>spring-cloud-stream-rxjava</module>
 		<module>spring-cloud-stream-samples</module>
 		<module>spring-cloud-stream-module-launcher</module>
 		<module>docs</module>
@@ -139,6 +141,11 @@
 				<artifactId>kryo-shaded</artifactId>
 				<version>3.0.3</version>
 			</dependency>
+			<dependency>
+				<groupId>io.reactivex</groupId>
+				<artifactId>rxjava</artifactId>
+				<version>${rx-java.version}</version>
+			</dependency>
 			<dependency>
 				<groupId>org.objenesis</groupId>
 				<artifactId>objenesis</artifactId>

spring-cloud-stream-rxjava/pom.xml

@@ -0,0 +1,32 @@
+<?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>
+
+	<artifactId>spring-cloud-stream-rxjava</artifactId>
+	<packaging>jar</packaging>
+	<name>spring-cloud-stream-rxjava</name>
+	<description>RxJava support for spring cloud stream modules</description>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-stream-parent</artifactId>
+		<version>1.0.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-core</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>io.reactivex</groupId>
+			<artifactId>rxjava</artifactId>
+		</dependency>
+	</dependencies>
+</project>

spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/EnableRxJavaProcessor.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.cloud.stream.annotation.rxjava;
+
+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.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.context.annotation.Import;
+
+/**
+ * Annotation that identifies the class as RxJava processor module. The class that has {@link EnableRxJavaProcessor}
+ * annotated is expected to provide a bean that implements {@link RxJavaProcessor}.
+ *
+ * @author Ilayaperumal Gopinathan
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@Inherited
+@EnableBinding(Processor.class)
+@Import(RxJavaProcessorConfiguration.class)
+public @interface EnableRxJavaProcessor {
+}

spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/RxJavaProcessor.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.annotation.rxjava;
+
+import rx.Observable;
+
+/**
+ * Marker interface that RxJava processor module uses to provide the implementation bean.
+ *
+ * @author Mark Pollack
+ * @author Ilayaperumal Gopinathan
+ */
+public interface RxJavaProcessor<I, O> {
+
+	Observable<O> process(Observable<I> input);
+}

spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/RxJavaProcessorConfiguration.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.cloud.stream.annotation.rxjava;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.messaging.MessageHandler;
+
+/**
+ * Configuration class for RxJava module support.
+ *
+ * @author Ilayaperumal Gopinathan
+ */
+@Configuration
+public class RxJavaProcessorConfiguration {
+
+	@Autowired
+	RxJavaProcessor processor;
+
+	@ServiceActivator(inputChannel = Processor.INPUT)
+	@Bean
+	public MessageHandler subjectMessageHandler() {
+		SubjectMessageHandler messageHandler = new SubjectMessageHandler(processor);
+		messageHandler.setOutputChannelName(Processor.OUTPUT);
+		return messageHandler;
+	}
+}

spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/SubjectMessageHandler.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.annotation.rxjava;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+
+import rx.Observable;
+import rx.Subscription;
+import rx.functions.Action0;
+import rx.functions.Action1;
+import rx.subjects.PublishSubject;
+import rx.subjects.SerializedSubject;
+import rx.subjects.Subject;
+
+/**
+ * Adapts the item at a time delivery of a {@link org.springframework.messaging.MessageHandler}
+ * by delegating processing to a {@link Observable}.
+ * <p/>
+ * The outputStream of the processor is used to create a message and send it to the output channel.  If the
+ * input channel and output channel are connected to the {@link org.springframework.cloud.stream.binder.Binder},
+ * then data delivered to the input stream via a call to onNext is invoked on the dispatcher thread of the binder
+ * and sending a message to the output channel will involve IO operations on the binder.
+ * <p/>
+ * The implementation uses a SerializedSubject.  This has the advantage that the state of the Observabale
+ * can be shared across all the incoming dispatcher threads that are invoking onNext.  It has the disadvantage
+ * that processing and sending to the output channel will execute serially on one of the dispatcher threads.
+ * <p/>
+ * The use of this handler makes for a very natural first experience when processing data.  For example given
+ * the stream <code></code>http | rxjava-processor | log</code> where the <code>rxjava-processor</code> does a
+ * <code>buffer(5)</code> and then produces a single value.  Sending 10 messages to the http source will
+ * result in 2 messages in the log, no matter how many dispatcher threads are used.
+ * <p/>
+ * You can modify what thread the outputStream subscriber, which does the send to the output channel,
+ * will use by explicitly calling <code>observeOn</code> before returning the outputStream from your processor.
+ * <p/>
+
+ * All error handling is the responsibility of the processor implementation.
+ *
+ * @author Mark Pollack
+ * @author Ilayaperumal Gopinathan
+ */
+@SuppressWarnings({"unchecked", "rawtypes"})
+public class SubjectMessageHandler extends AbstractMessageProducingHandler implements DisposableBean {
+
+	private final Logger logger = LoggerFactory.getLogger(getClass());
+
+	@SuppressWarnings("rawtypes")
+	private final RxJavaProcessor processor;
+
+	private final Subject subject;
+
+	private final Subscription subscription;
+
+	@SuppressWarnings({"unchecked", "rawtypes"})
+	public SubjectMessageHandler(RxJavaProcessor processor) {
+		Assert.notNull(processor, "RxJava processor must not be null.");
+		this.processor = processor;
+		subject = new SerializedSubject(PublishSubject.create());
+		Observable<?> outputStream = processor.process(subject);
+		subscription = outputStream.subscribe(new Action1<Object>() {
+			@Override
+			public void call(Object outputObject) {
+				if (ClassUtils.isAssignable(Message.class, outputObject.getClass())) {
+					getOutputChannel().send((Message) outputObject);
+				}
+				else {
+					getOutputChannel().send(MessageBuilder.withPayload(outputObject).build());
+				}
+			}
+		}, new Action1<Throwable>() {
+			@Override
+			public void call(Throwable throwable) {
+				logger.error(throwable.getMessage(), throwable);
+			}
+		}, new Action0() {
+			@Override
+			public void call() {
+				logger.error("Subscription close for [" + subscription + "]");
+			}
+		});
+	}
+
+
+	@Override
+	//todo: support module input type
+	protected void handleMessageInternal(Message<?> message) throws Exception {
+		subject.onNext(message.getPayload());
+	}
+
+	@Override
+	public void destroy() throws Exception {
+		subscription.unsubscribe();
+	}
+}

spring-cloud-stream-samples/pom.xml

@@ -25,6 +25,7 @@
 		<module>tap</module>
 		<module>double</module>
 		<module>extended</module>
+		<module>rxjava-processor</module>
 	</modules>
 	<dependencyManagement>
 		<dependencies>

spring-cloud-stream-samples/rxjava-processor/pom.xml

@@ -0,0 +1,63 @@
+<?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>
+
+	<artifactId>spring-cloud-stream-sample-rxjava</artifactId>
+	<packaging>jar</packaging>
+
+	<name>spring-cloud-stream-sample-rxjava</name>
+	<description>Demo project for RxJava module</description>
+
+	<parent>
+		<groupId>org.springframework.cloud</groupId>
+		<artifactId>spring-cloud-stream-samples</artifactId>
+		<version>1.0.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+		<start-class>demo.RxJavaApplication</start-class>
+		<java.version>1.8</java.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-rxjava</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-binder-redis</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-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>
+				<configuration>
+					<classifier>exec</classifier>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
+
+</project>

spring-cloud-stream-samples/rxjava-processor/src/main/java/demo/RxJavaApplication.java

@@ -0,0 +1,33 @@
+/*
+ * 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 demo;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.ComponentScan;
+
+/**
+ * @author Ilayaperumal Gopinathan
+ */
+@SpringBootApplication
+public class RxJavaApplication {
+
+	public static void main(String[] args) {
+		SpringApplication.run(RxJavaApplication.class, args);
+	}
+
+}

spring-cloud-stream-samples/rxjava-processor/src/main/java/demo/RxJavaTransformer.java

@@ -0,0 +1,51 @@
+/*
+ * 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 demo;
+
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.cloud.stream.annotation.rxjava.EnableRxJavaProcessor;
+import org.springframework.cloud.stream.annotation.rxjava.RxJavaProcessor;
+import org.springframework.context.annotation.Bean;
+
+@EnableRxJavaProcessor
+public class RxJavaTransformer {
+
+	private static Logger logger = LoggerFactory.getLogger(RxJavaTransformer.class);
+
+	@Bean
+	public RxJavaProcessor<String,String> processor() {
+		return inputStream -> inputStream.map(data -> {
+			logger.info("Got data = " + data);
+			return data;
+		}).buffer(5).map(data -> String.valueOf(avg(data)));
+	}
+
+	private static Double avg(List<String> data) {
+		double sum = 0;
+		double count = 0;
+		for(String d : data) {
+			count++;
+			sum += Double.valueOf(d);
+		}
+		return sum/count;
+	}
+
+}

spring-cloud-stream-samples/rxjava-processor/src/main/resources/application.yml

@@ -0,0 +1,8 @@
+server:
+  port: 8082
+spring:
+  cloud:
+    stream:
+      bindings:
+        output: xformed
+        input: testtock

spring-cloud-stream-samples/rxjava-processor/src/test/java/demo/ModuleApplicationTests.java

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

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java

@@ -38,7 +38,7 @@ import org.springframework.integration.config.EnableIntegration;
  * @author Marius Bogoevici
  * @author David Turanski
  */
-@Target(ElementType.TYPE)
+@Target({ElementType.TYPE, ElementType.ANNOTATION_TYPE})
 @Retention(RetentionPolicy.RUNTIME)
 @Documented
 @Inherited