diff --git a/pom.xml b/pom.xml index b8948813e..619f7e9df 100644 --- a/pom.xml +++ b/pom.xml @@ -37,11 +37,6 @@ spring-cloud-stream-tools ${project.version} - - org.springframework.cloud - spring-cloud-stream-rxjava - ${project.version} - org.springframework.cloud spring-cloud-stream-schema-server diff --git a/spring-cloud-stream-reactive/pom.xml b/spring-cloud-stream-reactive/pom.xml index b101443d6..c8d305df5 100644 --- a/spring-cloud-stream-reactive/pom.xml +++ b/spring-cloud-stream-reactive/pom.xml @@ -18,15 +18,6 @@ io.projectreactor reactor-core - - io.reactivex - rxjava - true - - - io.reactivex - rxjava-reactive-streams - org.springframework.boot spring-boot-starter-test @@ -44,39 +35,4 @@ - - - - org.apache.maven.plugins - maven-shade-plugin - 2.4.3 - - - package - - shade - - - - - io.reactivex:rxjava-reactive-streams - - - - - rx.RxReactiveStreams - org.springframework.cloud.stream.reactive.shaded.rx.RxReactiveStreams - - - rx.internal.reactivestreams - org.springframework.cloud.stream.reactive.shaded.rx.internal.reactivestreams - - - - - - - - - diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputObservableParameterAdapter.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputObservableParameterAdapter.java deleted file mode 100644 index ef5f99865..000000000 --- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputObservableParameterAdapter.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * Copyright 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.reactive; - -import rx.Observable; -import rx.RxReactiveStreams; - -import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter; -import org.springframework.core.MethodParameter; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.SubscribableChannel; -import org.springframework.util.Assert; - -/** - * Adapts an {@link org.springframework.cloud.stream.annotation.Input} annotated - * {@link MessageChannel} to an {@link Observable}. - * - * @deprecated as of 2.0.0.RELEASE, this is deprecated and will be removed in 2.1 Use Project Reactor based alternatives offered in Spring Cloud Stream. - * - * @author Marius Bogoevici - */ -public class MessageChannelToInputObservableParameterAdapter - implements StreamListenerParameterAdapter, SubscribableChannel> { - - private final MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxArgumentAdapter; - - public MessageChannelToInputObservableParameterAdapter( - MessageChannelToInputFluxParameterAdapter messageChannelToInputFluxArgumentAdapter) { - Assert.notNull(messageChannelToInputFluxArgumentAdapter, "cannot be null"); - this.messageChannelToInputFluxArgumentAdapter = messageChannelToInputFluxArgumentAdapter; - } - - @Override - public boolean supports(Class bindingTargetType, MethodParameter methodParameter) { - return MessageChannel.class.isAssignableFrom(bindingTargetType) - && Observable.class.isAssignableFrom(methodParameter.getParameterType()); - } - - @Override - public Observable adapt(final SubscribableChannel bindingTarget, MethodParameter parameter) { - return RxReactiveStreams.toObservable( - this.messageChannelToInputFluxArgumentAdapter.adapt(bindingTarget, parameter)); - } -} diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToObservableSenderParameterAdapter.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToObservableSenderParameterAdapter.java deleted file mode 100644 index f228c7d8d..000000000 --- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToObservableSenderParameterAdapter.java +++ /dev/null @@ -1,73 +0,0 @@ -/* - * Copyright 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.reactive; - -import org.reactivestreams.Publisher; -import reactor.core.publisher.Flux; -import rx.Observable; -import rx.RxReactiveStreams; -import rx.Single; - -import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter; -import org.springframework.core.MethodParameter; -import org.springframework.core.ResolvableType; -import org.springframework.messaging.MessageChannel; -import org.springframework.util.Assert; - -/** - * Adapts an {@link org.springframework.cloud.stream.annotation.Output} annotated - * {@link ObservableSender} to an outbound {@link MessageChannel}. - * - * @deprecated as of 2.0.0.RELEASE, this is deprecated and will be removed in 2.1 Use Project Reactor based alternatives offered in Spring Cloud Stream. - * - * @author Marius Bogoevici - */ -@Deprecated -public class MessageChannelToObservableSenderParameterAdapter implements - StreamListenerParameterAdapter { - - private final MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter; - - public MessageChannelToObservableSenderParameterAdapter( - MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter) { - Assert.notNull(messageChannelToFluxSenderArgumentAdapter, "cannot be null"); - this.messageChannelToFluxSenderArgumentAdapter = messageChannelToFluxSenderArgumentAdapter; - } - - @Override - public boolean supports(Class bindingTargetType, MethodParameter methodParameter) { - ResolvableType type = ResolvableType.forMethodParameter(methodParameter); - return MessageChannel.class.isAssignableFrom(bindingTargetType) - && ObservableSender.class.isAssignableFrom(type.getRawClass()); - } - - @Override - public ObservableSender adapt(MessageChannel bindingTarget, MethodParameter parameter) { - return new ObservableSender() { - - private FluxSender fluxSender = MessageChannelToObservableSenderParameterAdapter.this.messageChannelToFluxSenderArgumentAdapter - .adapt(bindingTarget, parameter); - - @Override - public Single send(Observable observable) { - Publisher adaptedPublisher = RxReactiveStreams.toPublisher(observable); - return RxReactiveStreams.toSingle( - this.fluxSender.send(Flux.from(adaptedPublisher))); - } - }; - } -} diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableSender.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableSender.java deleted file mode 100644 index aa6c4114a..000000000 --- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableSender.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Copyright 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.reactive; - -import rx.Observable; -import rx.Single; - -/** - * Used for {@link org.springframework.cloud.stream.annotation.StreamListener} arguments - * annotated with {@link org.springframework.cloud.stream.annotation.Output}. - * - * @deprecated as of 2.0.0.RELEASE, this interface is deprecated and will be removed in 2.1 Use Project Reactor based {@link FluxSender}. - * - * @author Marius Bogoevici - */ -@Deprecated -public interface ObservableSender { - - /** - * Streams the {@link Observable} through the binding target corresponding to the - * {@link org.springframework.cloud.stream.annotation.Output} annotation of the - * argument. - * - * @param observable an {@link Observable} that will be streamed through the bound - * element - * @return a {@link Single} representing the result of an operation - */ - Single send(Observable observable); -} diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableToMessageChannelResultAdapter.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableToMessageChannelResultAdapter.java deleted file mode 100644 index 0f57b0404..000000000 --- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableToMessageChannelResultAdapter.java +++ /dev/null @@ -1,59 +0,0 @@ -/* - * Copyright 2016-2017 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.reactive; - -import java.io.Closeable; - -import org.reactivestreams.Publisher; -import rx.Observable; -import rx.RxReactiveStreams; - -import org.springframework.cloud.stream.binding.StreamListenerResultAdapter; -import org.springframework.messaging.MessageChannel; -import org.springframework.util.Assert; - -/** - * A {@link StreamListenerResultAdapter} from an {@link Observable} return type to a bound - * {@link MessageChannel}. - * - * @author Marius Bogoevici - * @deprecated as of 2.0.0.RELEASE, this is deprecated and will be removed in 2.1 Use Project Reactor based alternatives offered in Spring Cloud Stream. - */ -@Deprecated -public class ObservableToMessageChannelResultAdapter - implements StreamListenerResultAdapter, MessageChannel> { - - private PublisherToMessageChannelResultAdapter publisherToMessageChannelResultAdapter; - - public ObservableToMessageChannelResultAdapter( - PublisherToMessageChannelResultAdapter publisherToMessageChannelResultAdapter) { - Assert.notNull(publisherToMessageChannelResultAdapter, "cannot be null"); - this.publisherToMessageChannelResultAdapter = publisherToMessageChannelResultAdapter; - } - - @Override - public boolean supports(Class resultType, Class bindingTarget) { - return Observable.class.isAssignableFrom(resultType) - && MessageChannel.class.isAssignableFrom(bindingTarget); - } - - public Closeable adapt(Observable streamListenerResult, MessageChannel bindingTarget) { - Publisher adaptedPublisher = RxReactiveStreams.toPublisher(streamListenerResult); - return this.publisherToMessageChannelResultAdapter.adapt(adaptedPublisher, bindingTarget); - } - -} diff --git a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java index 92f1b5835..81f9c005c 100644 --- a/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java +++ b/spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java @@ -17,7 +17,6 @@ package org.springframework.cloud.stream.reactive; import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.cloud.stream.binding.BindingService; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; @@ -55,30 +54,4 @@ public class ReactiveSupportAutoConfiguration { public static StreamEmitterAnnotationBeanPostProcessor streamEmitterAnnotationBeanPostProcessor() { return new StreamEmitterAnnotationBeanPostProcessor(); } - - @Configuration - @ConditionalOnClass(name = "rx.Observable") - public static class RxJava1SupportConfiguration { - - @Bean - @ConditionalOnMissingBean(MessageChannelToInputObservableParameterAdapter.class) - public MessageChannelToInputObservableParameterAdapter messageChannelToInputObservableArgumentAdapter( - MessageChannelToInputFluxParameterAdapter messageChannelToFluxArgumentAdapter) { - return new MessageChannelToInputObservableParameterAdapter(messageChannelToFluxArgumentAdapter); - } - - @Bean - @ConditionalOnMissingBean(MessageChannelToObservableSenderParameterAdapter.class) - public MessageChannelToObservableSenderParameterAdapter messageChannelToObservableSenderArgumentAdapter( - MessageChannelToFluxSenderParameterAdapter messageChannelToFluxSenderArgumentAdapter) { - return new MessageChannelToObservableSenderParameterAdapter(messageChannelToFluxSenderArgumentAdapter); - } - - @Bean - @ConditionalOnMissingBean(ObservableToMessageChannelResultAdapter.class) - public ObservableToMessageChannelResultAdapter observableToMessageChannelResultAdapter( - PublisherToMessageChannelResultAdapter publisherToMessageChannelResultAdapter) { - return new ObservableToMessageChannelResultAdapter(publisherToMessageChannelResultAdapter); - } - } } diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java index eb73f2e1b..9e9a4456f 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java @@ -16,8 +16,8 @@ package org.springframework.cloud.stream.reactive; -import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -25,7 +25,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; -import rx.Observable; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -57,7 +56,7 @@ public class StreamListenerReactiveInputOutputArgsTests { @Parameterized.Parameters public static Collection InputConfigs() { - return Arrays.asList(new Class[] { ReactorTestInputOutputArgs.class, RxJava1TestInputOutputArgs.class }); + return Collections.singletonList(ReactorTestInputOutputArgs.class); } @SuppressWarnings("unchecked") @@ -90,15 +89,4 @@ public class StreamListenerReactiveInputOutputArgsTests { output.send(input.map(m -> m.toUpperCase())); } } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestInputOutputArgs { - - @StreamListener - public void receive(@Input(Processor.INPUT) Observable input, - @Output(Processor.OUTPUT) ObservableSender output) { - output.send(input.map(m -> m.toUpperCase())); - } - } } diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java index 30cb3a651..afe86006c 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java @@ -16,8 +16,8 @@ package org.springframework.cloud.stream.reactive; -import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -25,7 +25,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; -import rx.Observable; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -39,6 +38,7 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; + import static org.assertj.core.api.Assertions.assertThat; /** @@ -57,8 +57,7 @@ public class StreamListenerReactiveInputOutputArgsWithMessageTests { @Parameterized.Parameters public static Collection InputConfigs() { - return Arrays.asList(new Class[] { ReactorTestInputOutputArgsWithMessage.class, - RxJava1TestInputOutputArgsWithMessage.class }); + return Collections.singletonList(ReactorTestInputOutputArgsWithMessage.class); } @SuppressWarnings("unchecked") @@ -93,15 +92,4 @@ public class StreamListenerReactiveInputOutputArgsWithMessageTests { .withPayload(m.getPayload().toString().toUpperCase()).build())); } } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestInputOutputArgsWithMessage { - - @StreamListener - public void receive(@Input(Processor.INPUT) Observable> input, - @Output(Processor.OUTPUT) ObservableSender output) { - output.send(input.map(m -> MessageBuilder.withPayload(m.getPayload().toUpperCase()).build())); - } - } } diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java index 02587137a..a51516adc 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java @@ -16,8 +16,8 @@ package org.springframework.cloud.stream.reactive; -import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -25,7 +25,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; -import rx.Observable; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -39,6 +38,7 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; + import static org.assertj.core.api.Assertions.assertThat; /** @@ -57,8 +57,7 @@ public class StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests { @Parameterized.Parameters public static Collection InputConfigs() { - return Arrays.asList(new Class[] { TestInputOutputArgsWithFluxSenderAndFailure.class, - TestInputOutputArgsWithObservableSenderAndFailure.class }); + return Collections.singletonList(TestInputOutputArgsWithFluxSenderAndFailure.class); } @SuppressWarnings("unchecked") @@ -108,24 +107,4 @@ public class StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests { .map(o -> MessageBuilder.withPayload(o).build())); } } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class TestInputOutputArgsWithObservableSenderAndFailure { - @StreamListener - public void receive(@Input(Processor.INPUT) Observable> input, - @Output(Processor.OUTPUT) ObservableSender output) { - output.send(input - .map(m -> m.getPayload().toString()) - .map(m -> { - if (!m.equals("fail")) { - return m.toUpperCase(); - } - else { - throw new RuntimeException(); - } - }) - .map(o -> MessageBuilder.withPayload(o).build())); - } - } } diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java index 8312da83c..6d5fe6fdf 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java @@ -16,8 +16,8 @@ package org.springframework.cloud.stream.reactive; -import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -25,7 +25,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; -import rx.Observable; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -57,8 +56,7 @@ public class StreamListenerReactiveInputOutputArgsWithSenderTests { @Parameterized.Parameters public static Collection InputConfigs() { - return Arrays.asList(new Class[] { ReactorTestInputOutputArgsWithFluxSender.class, - RxJava1TestInputOutputArgsWithObservableSender.class }); + return Collections.singletonList(ReactorTestInputOutputArgsWithFluxSender.class); } @SuppressWarnings("unchecked") @@ -97,16 +95,4 @@ public class StreamListenerReactiveInputOutputArgsWithSenderTests { .map(o -> MessageBuilder.withPayload(o).build())); } } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestInputOutputArgsWithObservableSender { - @StreamListener - public void receive(@Input(Processor.INPUT) Observable> input, - @Output(Processor.OUTPUT) ObservableSender output) { - output.send(input - .map(m -> m.getPayload().toString().toUpperCase()) - .map(o -> MessageBuilder.withPayload(o).build())); - } - } } diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java index 8f84d4c7c..956517b95 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java @@ -18,7 +18,6 @@ package org.springframework.cloud.stream.reactive; import org.junit.Test; import reactor.core.publisher.Flux; -import rx.Observable; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -49,20 +48,6 @@ public class StreamListenerReactiveMethodTests { } } - @Test - public void testRxJava1InvalidInputValueWithOutputMethodParameters() { - try { - SpringApplication.run(RxJava1TestInputOutputArgs.class, "--server.port=0", - "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.input.contentType=text/plain", - "--spring.cloud.stream.bindings.output.contentType=text/plain"); - fail("IllegalArgumentException should have been thrown"); - } - catch (Exception e) { - assertThat(e.getMessage()).contains(INVALID_INPUT_VALUE_WITH_OUTPUT_METHOD_PARAM); - } - } - @Test public void testMethodReturnTypeWithNoOutboundSpecified() { try { @@ -87,16 +72,6 @@ public class StreamListenerReactiveMethodTests { } } - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestInputOutputArgs { - - @StreamListener(Processor.INPUT) - public void receive(Observable input, @Output(Processor.OUTPUT) ObservableSender output) { - output.send(input.map(m -> m.toUpperCase())); - } - } - @EnableBinding(Processor.class) @EnableAutoConfiguration public static class ReactorTestReturn5 { diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java index d5b4b0379..79e744b3c 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java @@ -25,7 +25,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; -import rx.Observable; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -58,10 +57,8 @@ public class StreamListenerReactiveMethodWithReturnTypeTests { @Parameterized.Parameters public static Collection InputConfigs() { - return Arrays.asList(new Class[] { ReactorTestReturn1.class, ReactorTestReturn2.class, ReactorTestReturn3.class, - ReactorTestReturn4.class, - RxJava1TestReturn1.class, RxJava1TestReturn2.class, RxJava1TestReturn3.class, - RxJava1TestReturn4.class }); + return Arrays.asList(ReactorTestReturn1.class, ReactorTestReturn2.class, ReactorTestReturn3.class, + ReactorTestReturn4.class); } @SuppressWarnings("unchecked") @@ -129,44 +126,4 @@ public class StreamListenerReactiveMethodWithReturnTypeTests { return input.map(m -> m.toUpperCase()); } } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturn1 { - - @StreamListener - public @Output(Processor.OUTPUT) Observable receive(@Input(Processor.INPUT) Observable input) { - return input.map(m -> m.toUpperCase()); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturn2 { - - @StreamListener(Processor.INPUT) - public @Output(Processor.OUTPUT) Observable receive(Observable input) { - return input.map(m -> m.toUpperCase()); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturn3 { - - @StreamListener(Processor.INPUT) - public @SendTo(Processor.OUTPUT) Observable receive(Observable input) { - return input.map(m -> m.toUpperCase()); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturn4 { - - @StreamListener - public @SendTo(Processor.OUTPUT) Observable receive(@Input(Processor.INPUT) Observable input) { - return input.map(m -> m.toUpperCase()); - } - } } diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java index 6b97702cf..073bd525c 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java @@ -25,7 +25,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; -import rx.Observable; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -58,11 +57,8 @@ public class StreamListenerReactiveReturnWithFailureTests { @Parameterized.Parameters public static Collection InputConfigs() { - return Arrays.asList(new Class[] { ReactorTestReturnWithFailure1.class, ReactorTestReturnWithFailure2.class, - ReactorTestReturnWithFailure3.class, ReactorTestReturnWithFailure4.class, - RxJava1TestReturnWithFailure1.class, - RxJava1TestReturnWithFailure2.class, RxJava1TestReturnWithFailure3.class, - RxJava1TestReturnWithFailure4.class }); + return Arrays.asList(ReactorTestReturnWithFailure1.class, ReactorTestReturnWithFailure2.class, + ReactorTestReturnWithFailure3.class, ReactorTestReturnWithFailure4.class); } @SuppressWarnings("unchecked") @@ -76,7 +72,7 @@ public class StreamListenerReactiveReturnWithFailureTests { assertThat(result.getPayload()).isEqualTo(sentPayload.toUpperCase()); } - private static void sendFailingMessage(ConfigurableApplicationContext context) throws InterruptedException { + private static void sendFailingMessage(ConfigurableApplicationContext context) { Processor processor = context.getBean(Processor.class); processor.input().send(MessageBuilder.withPayload("fail").setHeader("contentType", "text/plain").build()); } @@ -100,12 +96,12 @@ public class StreamListenerReactiveReturnWithFailureTests { public static class ReactorTestReturnWithFailure1 { @StreamListener - public @Output(Processor.OUTPUT) Flux receive(@Input(Processor.INPUT) Flux input) { + public @Output(Processor.OUTPUT) + Flux receive(@Input(Processor.INPUT) Flux input) { return input.map(m -> { if (!m.equals("fail")) { return m.toUpperCase(); - } - else { + } else { throw new RuntimeException(); } }); @@ -117,12 +113,12 @@ public class StreamListenerReactiveReturnWithFailureTests { public static class ReactorTestReturnWithFailure2 { @StreamListener(Processor.INPUT) - public @Output(Processor.OUTPUT) Flux receive(Flux input) { + public @Output(Processor.OUTPUT) + Flux receive(Flux input) { return input.map(m -> { if (!m.equals("fail")) { return m.toUpperCase(); - } - else { + } else { throw new RuntimeException(); } }); @@ -134,12 +130,12 @@ public class StreamListenerReactiveReturnWithFailureTests { public static class ReactorTestReturnWithFailure3 { @StreamListener(Processor.INPUT) - public @SendTo(Processor.OUTPUT) Flux receive(Flux input) { + public @SendTo(Processor.OUTPUT) + Flux receive(Flux input) { return input.map(m -> { if (!m.equals("fail")) { return m.toUpperCase(); - } - else { + } else { throw new RuntimeException(); } }); @@ -151,80 +147,12 @@ public class StreamListenerReactiveReturnWithFailureTests { public static class ReactorTestReturnWithFailure4 { @StreamListener - public @SendTo(Processor.OUTPUT) Flux receive(@Input(Processor.INPUT) Flux input) { + public @SendTo(Processor.OUTPUT) + Flux receive(@Input(Processor.INPUT) Flux input) { return input.map(m -> { if (!m.equals("fail")) { return m.toUpperCase(); - } - else { - throw new RuntimeException(); - } - }); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithFailure1 { - - @StreamListener - public @Output(Processor.OUTPUT) Observable receive(@Input(Processor.INPUT) Observable input) { - return input.map(m -> { - if (!m.equals("fail")) { - return m.toUpperCase(); - } - else { - throw new RuntimeException(); - } - }); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithFailure2 { - - @StreamListener - public @SendTo(Processor.OUTPUT) Observable receive(@Input(Processor.INPUT) Observable input) { - return input.map(m -> { - if (!m.equals("fail")) { - return m.toUpperCase(); - } - else { - throw new RuntimeException(); - } - }); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithFailure3 { - - @StreamListener(Processor.INPUT) - public @SendTo(Processor.OUTPUT) Observable receive(Observable input) { - return input.map(m -> { - if (!m.equals("fail")) { - return m.toUpperCase(); - } - else { - throw new RuntimeException(); - } - }); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithFailure4 { - - @StreamListener(Processor.INPUT) - public @Output(Processor.OUTPUT) Observable receive(Observable input) { - return input.map(m -> { - if (!m.equals("fail")) { - return m.toUpperCase(); - } - else { + } else { throw new RuntimeException(); } }); diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java index cc8438c17..cf5bb5a20 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java @@ -25,7 +25,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; -import rx.Observable; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -58,11 +57,8 @@ public class StreamListenerReactiveReturnWithMessageTests { @Parameterized.Parameters public static Collection InputConfigs() { - return Arrays.asList(new Class[] { ReactorTestReturnWithMessage1.class, ReactorTestReturnWithMessage2.class, - ReactorTestReturnWithMessage3.class, ReactorTestReturnWithMessage4.class, - RxJava1TestReturnWithMessage1.class, - RxJava1TestReturnWithMessage2.class, RxJava1TestReturnWithMessage3.class, - RxJava1TestReturnWithMessage4.class }); + return Arrays.asList(ReactorTestReturnWithMessage1.class, ReactorTestReturnWithMessage2.class, + ReactorTestReturnWithMessage3.class, ReactorTestReturnWithMessage4.class); } @SuppressWarnings("unchecked") @@ -125,47 +121,4 @@ public class StreamListenerReactiveReturnWithMessageTests { return input.map(m -> m.getPayload().toUpperCase()); } } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithMessage1 { - - @StreamListener - public @Output(Processor.OUTPUT) Observable receive( - @Input(Processor.INPUT) Observable> input) { - return input.map(m -> m.getPayload().toUpperCase()); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithMessage2 { - - @StreamListener - public @SendTo(Processor.OUTPUT) Observable receive( - @Input(Processor.INPUT) Observable> input) { - return input.map(m -> m.getPayload().toUpperCase()); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithMessage3 { - - @StreamListener(Processor.INPUT) - public @Output(Processor.OUTPUT) Observable receive(Observable> input) { - return input.map(m -> m.getPayload().toUpperCase()); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithMessage4 { - - @StreamListener(Processor.INPUT) - public @SendTo(Processor.OUTPUT) Observable receive(Observable> input) { - return input.map(m -> m.getPayload().toUpperCase()); - } - } - } diff --git a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java index 7ec73fe78..94d010863 100644 --- a/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java +++ b/spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java @@ -27,7 +27,6 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Parameterized; import reactor.core.publisher.Flux; -import rx.Observable; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -62,9 +61,8 @@ public class StreamListenerReactiveReturnWithPojoTests { @Parameterized.Parameters public static Collection InputConfigs() { - return Arrays.asList(new Class[] { ReactorTestReturnWithPojo1.class, ReactorTestReturnWithPojo2.class, - ReactorTestReturnWithPojo3.class, ReactorTestReturnWithPojo4.class, RxJava1TestReturnWithPojo1.class, - RxJava1TestReturnWithPojo2.class, RxJava1TestReturnWithPojo3.class, RxJava1TestReturnWithPojo4.class }); + return Arrays.asList(ReactorTestReturnWithPojo1.class, ReactorTestReturnWithPojo2.class, + ReactorTestReturnWithPojo3.class, ReactorTestReturnWithPojo4.class); } @Test @@ -123,48 +121,6 @@ public class StreamListenerReactiveReturnWithPojoTests { } } - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithPojo1 { - - @StreamListener - public @Output(Processor.OUTPUT) Observable receive( - @Input(Processor.INPUT) Observable input) { - return input.map(m -> new BarPojo(m.getMessage())); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithPojo2 { - - @StreamListener - public @SendTo(Processor.OUTPUT) Observable receive( - @Input(Processor.INPUT) Observable input) { - return input.map(m -> new BarPojo(m.getMessage())); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithPojo3 { - - @StreamListener(Processor.INPUT) - public @Output(Processor.OUTPUT) Observable receive(Observable input) { - return input.map(m -> new BarPojo(m.getMessage())); - } - } - - @EnableBinding(Processor.class) - @EnableAutoConfiguration - public static class RxJava1TestReturnWithPojo4 { - - @StreamListener(Processor.INPUT) - public @SendTo(Processor.OUTPUT) Observable receive(Observable input) { - return input.map(m -> new BarPojo(m.getMessage())); - } - } - public static class FooPojo { private String message;