SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove RxJava components from the reactive module

Browse Source 
Remove all the deprecated RxJava components from spring-cloud-stream-reactive module in favor
of using Project Reactor based types.

Fixes #1378
Resolves #1379
This commit is contained in:
Soby Chacko
2018-05-14 15:36:04 -04:00
committed by Oleg Zhurakousky
parent 66254ca9f6
commit 624a25fc92
16 changed files with 31 additions and 630 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
pom.xml
View File

@@ -37,11 +37,6 @@
<artifactId>spring-cloud-stream-tools</artifactId>
<version>${project.version}</version>
</dependency>
<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-schema-server</artifactId>

44
spring-cloud-stream-reactive/pom.xml
View File

@@ -18,15 +18,6 @@
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
</dependency>
<dependency>
<groupId>io.reactivex</groupId>
<artifactId>rxjava</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.reactivex</groupId>
<artifactId>rxjava-reactive-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
@@ -44,39 +35,4 @@
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.4.3</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<artifactSet>
<includes>
<include>io.reactivex:rxjava-reactive-streams</include>
</includes>
</artifactSet>
<relocations>
<relocation>
<pattern>rx.RxReactiveStreams</pattern>
<shadedPattern>org.springframework.cloud.stream.reactive.shaded.rx.RxReactiveStreams</shadedPattern>
</relocation>
<relocation>
<pattern>rx.internal.reactivestreams</pattern>
<shadedPattern>org.springframework.cloud.stream.reactive.shaded.rx.internal.reactivestreams</shadedPattern>
</relocation>
</relocations>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

58
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToInputObservableParameterAdapter.java
View File

@@ -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<Observable<?>, 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));
}
}

73
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/MessageChannelToObservableSenderParameterAdapter.java
View File

@@ -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<ObservableSender, MessageChannel> {
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<Void> send(Observable<?> observable) {
Publisher<?> adaptedPublisher = RxReactiveStreams.toPublisher(observable);
return RxReactiveStreams.toSingle(
this.fluxSender.send(Flux.from(adaptedPublisher)));
}
};
}
}

43
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableSender.java
View File

@@ -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<Void> send(Observable<?> observable);
}

59
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ObservableToMessageChannelResultAdapter.java
View File

@@ -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<Observable<?>, 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);
}
}

27
spring-cloud-stream-reactive/src/main/java/org/springframework/cloud/stream/reactive/ReactiveSupportAutoConfiguration.java
View File

@@ -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);
}
}
}

16
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsTests.java
View File

@@ -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<String> input,
@Output(Processor.OUTPUT) ObservableSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
}

18
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithMessageTests.java
View File

@@ -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<Message<String>> input,
@Output(Processor.OUTPUT) ObservableSender output) {
output.send(input.map(m -> MessageBuilder.withPayload(m.getPayload().toUpperCase()).build()));
}
}
}

27
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java
View File

@@ -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<Message<String>> 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()));
}
}
}

18
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveInputOutputArgsWithSenderTests.java
View File

@@ -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<Message<?>> input,
@Output(Processor.OUTPUT) ObservableSender output) {
output.send(input
.map(m -> m.getPayload().toString().toUpperCase())
.map(o -> MessageBuilder.withPayload(o).build()));
}
}
}

25
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodTests.java
View File

@@ -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<String> input, @Output(Processor.OUTPUT) ObservableSender output) {
output.send(input.map(m -> m.toUpperCase()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class ReactorTestReturn5 {

47
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveMethodWithReturnTypeTests.java
View File

@@ -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<String> receive(@Input(Processor.INPUT) Observable<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturn2 {
@StreamListener(Processor.INPUT)
public @Output(Processor.OUTPUT) Observable<String> receive(Observable<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturn3 {
@StreamListener(Processor.INPUT)
public @SendTo(Processor.OUTPUT) Observable<String> receive(Observable<String> input) {
return input.map(m -> m.toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturn4 {
@StreamListener
public @SendTo(Processor.OUTPUT) Observable<String> receive(@Input(Processor.INPUT) Observable<String> input) {
return input.map(m -> m.toUpperCase());
}
}
}

102
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithFailureTests.java
View File

@@ -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<String> receive(@Input(Processor.INPUT) Flux<String> input) {
public @Output(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<String> 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<String> receive(Flux<String> input) {
public @Output(Processor.OUTPUT)
Flux<String> receive(Flux<String> 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<String> receive(Flux<String> input) {
public @SendTo(Processor.OUTPUT)
Flux<String> receive(Flux<String> 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<String> receive(@Input(Processor.INPUT) Flux<String> input) {
public @SendTo(Processor.OUTPUT)
Flux<String> receive(@Input(Processor.INPUT) Flux<String> 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<String> receive(@Input(Processor.INPUT) Observable<String> 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<String> receive(@Input(Processor.INPUT) Observable<String> 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<String> receive(Observable<String> 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<String> receive(Observable<String> input) {
return input.map(m -> {
if (!m.equals("fail")) {
return m.toUpperCase();
}
else {
} else {
throw new RuntimeException();
}
});

51
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithMessageTests.java
View File

@@ -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<String> receive(
@Input(Processor.INPUT) Observable<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithMessage2 {
@StreamListener
public @SendTo(Processor.OUTPUT) Observable<String> receive(
@Input(Processor.INPUT) Observable<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithMessage3 {
@StreamListener(Processor.INPUT)
public @Output(Processor.OUTPUT) Observable<String> receive(Observable<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithMessage4 {
@StreamListener(Processor.INPUT)
public @SendTo(Processor.OUTPUT) Observable<String> receive(Observable<Message<String>> input) {
return input.map(m -> m.getPayload().toUpperCase());
}
}
}

48
spring-cloud-stream-reactive/src/test/java/org/springframework/cloud/stream/reactive/StreamListenerReactiveReturnWithPojoTests.java
View File

@@ -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<BarPojo> receive(
@Input(Processor.INPUT) Observable<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
@EnableBinding(Processor.class)
@EnableAutoConfiguration
public static class RxJava1TestReturnWithPojo2 {
@StreamListener
public @SendTo(Processor.OUTPUT) Observable<BarPojo> receive(
@Input(Processor.INPUT) Observable<FooPojo> 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<BarPojo> receive(Observable<FooPojo> 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<BarPojo> receive(Observable<FooPojo> input) {
return input.map(m -> new BarPojo(m.getMessage()));
}
}
public static class FooPojo {
private String message;