diff --git a/pom.xml b/pom.xml index 1edde76cc..f755e26a6 100644 --- a/pom.xml +++ b/pom.xml @@ -19,6 +19,7 @@ 1.7 + 1.0.14 1.3.0.BUILD-SNAPSHOT 4.2.1.BUILD-SNAPSHOT 4.2.0.BUILD-SNAPSHOT @@ -28,6 +29,7 @@ spring-cloud-stream-binders spring-cloud-stream-codec spring-cloud-stream-starters + spring-cloud-stream-rxjava spring-cloud-stream-samples spring-cloud-stream-module-launcher docs @@ -139,6 +141,11 @@ kryo-shaded 3.0.3 + + io.reactivex + rxjava + ${rx-java.version} + org.objenesis objenesis diff --git a/spring-cloud-stream-rxjava/pom.xml b/spring-cloud-stream-rxjava/pom.xml new file mode 100644 index 000000000..fd29bb1b2 --- /dev/null +++ b/spring-cloud-stream-rxjava/pom.xml @@ -0,0 +1,32 @@ + + + 4.0.0 + + spring-cloud-stream-rxjava + jar + spring-cloud-stream-rxjava + RxJava support for spring cloud stream modules + + + org.springframework.cloud + spring-cloud-stream-parent + 1.0.0.BUILD-SNAPSHOT + + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.integration + spring-integration-core + + + io.reactivex + rxjava + + + diff --git a/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/EnableRxJavaProcessor.java b/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/EnableRxJavaProcessor.java new file mode 100644 index 000000000..186359da8 --- /dev/null +++ b/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 { +} diff --git a/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/RxJavaProcessor.java b/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/RxJavaProcessor.java new file mode 100644 index 000000000..a10de2c86 --- /dev/null +++ b/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 { + + Observable process(Observable input); +} diff --git a/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/RxJavaProcessorConfiguration.java b/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/RxJavaProcessorConfiguration.java new file mode 100644 index 000000000..7524839d3 --- /dev/null +++ b/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; + } +} diff --git a/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/SubjectMessageHandler.java b/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/SubjectMessageHandler.java new file mode 100644 index 000000000..2c3a90c11 --- /dev/null +++ b/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}. + *

+ * 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. + *

+ * 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. + *

+ * The use of this handler makes for a very natural first experience when processing data. For example given + * the stream http | rxjava-processor | log where the rxjava-processor does a + * buffer(5) 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. + *

+ * You can modify what thread the outputStream subscriber, which does the send to the output channel, + * will use by explicitly calling observeOn before returning the outputStream from your processor. + *

+ + * 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() { + @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() { + @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(); + } +} diff --git a/spring-cloud-stream-samples/pom.xml b/spring-cloud-stream-samples/pom.xml index 4d208c965..ea139f5d6 100644 --- a/spring-cloud-stream-samples/pom.xml +++ b/spring-cloud-stream-samples/pom.xml @@ -25,6 +25,7 @@ tap double extended + rxjava-processor diff --git a/spring-cloud-stream-samples/rxjava-processor/pom.xml b/spring-cloud-stream-samples/rxjava-processor/pom.xml new file mode 100644 index 000000000..8fa7fcdec --- /dev/null +++ b/spring-cloud-stream-samples/rxjava-processor/pom.xml @@ -0,0 +1,63 @@ + + + 4.0.0 + + spring-cloud-stream-sample-rxjava + jar + + spring-cloud-stream-sample-rxjava + Demo project for RxJava module + + + org.springframework.cloud + spring-cloud-stream-samples + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + demo.RxJavaApplication + 1.8 + + + + + org.springframework.cloud + spring-cloud-stream-rxjava + ${project.version} + + + org.springframework.cloud + spring-cloud-stream-binder-redis + + + org.springframework.boot + spring-boot-starter-redis + + + org.springframework.boot + spring-boot-configuration-processor + true + + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + exec + + + + + + diff --git a/spring-cloud-stream-samples/rxjava-processor/src/main/java/demo/RxJavaApplication.java b/spring-cloud-stream-samples/rxjava-processor/src/main/java/demo/RxJavaApplication.java new file mode 100644 index 000000000..78fc2149d --- /dev/null +++ b/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); + } + +} diff --git a/spring-cloud-stream-samples/rxjava-processor/src/main/java/demo/RxJavaTransformer.java b/spring-cloud-stream-samples/rxjava-processor/src/main/java/demo/RxJavaTransformer.java new file mode 100644 index 000000000..8ed814677 --- /dev/null +++ b/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 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 data) { + double sum = 0; + double count = 0; + for(String d : data) { + count++; + sum += Double.valueOf(d); + } + return sum/count; + } + +} diff --git a/spring-cloud-stream-samples/rxjava-processor/src/main/resources/application.yml b/spring-cloud-stream-samples/rxjava-processor/src/main/resources/application.yml new file mode 100644 index 000000000..8668638ef --- /dev/null +++ b/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 diff --git a/spring-cloud-stream-samples/rxjava-processor/src/test/java/demo/ModuleApplicationTests.java b/spring-cloud-stream-samples/rxjava-processor/src/test/java/demo/ModuleApplicationTests.java new file mode 100644 index 000000000..4e6887ba4 --- /dev/null +++ b/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() { + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java index 4f423414a..2f94e08a0 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java +++ b/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