diff --git a/pom.xml b/pom.xml index a2fdbf889..03245a842 100644 --- a/pom.xml +++ b/pom.xml @@ -102,7 +102,6 @@ spring-cloud-stream spring-cloud-stream-binder-test - spring-cloud-stream-rxjava spring-cloud-stream-test-support spring-cloud-stream-test-support-internal spring-cloud-stream-integration-tests diff --git a/spring-cloud-stream-rxjava/.jdk8 b/spring-cloud-stream-rxjava/.jdk8 deleted file mode 100644 index e69de29bb..000000000 diff --git a/spring-cloud-stream-rxjava/pom.xml b/spring-cloud-stream-rxjava/pom.xml deleted file mode 100644 index 4dd174019..000000000 --- a/spring-cloud-stream-rxjava/pom.xml +++ /dev/null @@ -1,30 +0,0 @@ - - - 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 - 2.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 deleted file mode 100644 index a42edcf48..000000000 --- a/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/EnableRxJavaProcessor.java +++ /dev/null @@ -1,47 +0,0 @@ -/* - * 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 - * @deprecated in favor of - * {@link org.springframework.cloud.stream.annotation.StreamListener} with reactive types - */ -@Target(ElementType.TYPE) -@Retention(RetentionPolicy.RUNTIME) -@Documented -@Inherited -@EnableBinding(Processor.class) -@Import(RxJavaProcessorConfiguration.class) -@Deprecated -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 deleted file mode 100644 index bf7639ac4..000000000 --- a/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/RxJavaProcessor.java +++ /dev/null @@ -1,33 +0,0 @@ -/* - * 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 - * @deprecated in favor of - * {@link org.springframework.cloud.stream.annotation.StreamListener} with reactive types - */ -@Deprecated -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 deleted file mode 100644 index 30ac35329..000000000 --- a/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/RxJavaProcessorConfiguration.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Copyright 2015-2016 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 - * @author Marius Bogoevici - */ -@Configuration -@Deprecated -public class RxJavaProcessorConfiguration { - - @Autowired - RxJavaProcessor processor; - - @ServiceActivator(inputChannel = Processor.INPUT, phase = "0") - @Bean - public MessageHandler subjectMessageHandler() { - SubjectMessageHandler messageHandler = new SubjectMessageHandler(this.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 deleted file mode 100644 index 0c4d44a67..000000000 --- a/spring-cloud-stream-rxjava/src/main/java/org/springframework/cloud/stream/annotation/rxjava/SubjectMessageHandler.java +++ /dev/null @@ -1,164 +0,0 @@ -/* - * Copyright 2015-2016 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.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -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; - -import org.springframework.context.SmartLifecycle; -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; - -/** - * 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 - * @author Marius Bogoevici - */ -@SuppressWarnings({ "unchecked", "rawtypes" }) -@Deprecated -public class SubjectMessageHandler extends AbstractMessageProducingHandler implements SmartLifecycle { - - private final Log logger = LogFactory.getLog(getClass()); - - @SuppressWarnings("rawtypes") - private final RxJavaProcessor processor; - - private volatile Subject subject; - - private volatile Subscription subscription; - - private volatile boolean running; - - @SuppressWarnings({ "unchecked", "rawtypes" }) - public SubjectMessageHandler(RxJavaProcessor processor) { - Assert.notNull(processor, "RxJava processor must not be null."); - this.processor = processor; - } - - @Override - public synchronized void start() { - if (!this.running) { - this.subject = new SerializedSubject(PublishSubject.create()); - Observable outputStream = this.processor.process(this.subject); - this.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) { - SubjectMessageHandler.this.logger.error(throwable.getMessage(), throwable); - } - }, new Action0() { - - @Override - public void call() { - SubjectMessageHandler.this.logger - .info("Subscription close for [" + SubjectMessageHandler.this.subscription + "]"); - } - }); - this.running = true; - } - } - - @Override - public synchronized boolean isRunning() { - return this.running; - } - - @Override - public boolean isAutoStartup() { - return false; - } - - @Override - public void stop(Runnable callback) { - if (this.running) { - stop(); - if (callback != null) { - callback.run(); - } - } - } - - @Override - public int getPhase() { - return 0; - } - - @Override - protected void handleMessageInternal(Message message) throws Exception { - this.subject.onNext(message.getPayload()); - } - - @Override - public synchronized void stop() { - if (this.running) { - this.subject.onCompleted(); - this.subscription.unsubscribe(); - this.subscription = null; - this.subject = null; - this.running = false; - } - } -}