Updated RxJava transformer to 1.1.0

rxjava-processor/pom.xml

@@ -29,6 +29,20 @@
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-reactive</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-core</artifactId>
+			<version>3.0.1.BUILD-SNAPSHOT</version>
+		</dependency>
+		<dependency>
+			<groupId>io.reactivex</groupId>
+			<artifactId>rxjava</artifactId>
+			<version>1.1.10</version>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-redis</artifactId>

rxjava-processor/src/main/java/demo/RxJavaTransformer.java

@@ -20,19 +20,22 @@ import java.util.List;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import rx.Observable;
 
-import org.springframework.cloud.stream.annotation.rxjava.EnableRxJavaProcessor;
-import org.springframework.cloud.stream.annotation.rxjava.RxJavaProcessor;
-import org.springframework.context.annotation.Bean;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.messaging.handler.annotation.SendTo;
 
-@EnableRxJavaProcessor
+@EnableBinding(Processor.class)
 public class RxJavaTransformer {
 
 	private static Logger logger = LoggerFactory.getLogger(RxJavaTransformer.class);
 
-	@Bean
-	public RxJavaProcessor<String,String> processor() {
-		return inputStream -> inputStream.map(data -> {
+	@StreamListener(Processor.INPUT)
+	@SendTo(Processor.OUTPUT)
+	public Observable<String> processor(Observable<String> inputStream) {
+		return inputStream.map(data -> {
 			logger.info("Got data = " + data);
 			return data;
 		}).buffer(5).map(data -> String.valueOf(avg(data)));