Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add support for RxJava 3

Browse Source 
Closes gh-24170
This commit is contained in:
Rossen Stoyanchev
2020-05-11 08:40:01 +01:00
parent 3f30dcae80
commit 7f59381c7d
29 changed files with 450 additions and 500 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java
View File

@@ -24,8 +24,6 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.function.Function;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import kotlinx.coroutines.CompletableDeferredKt;
import kotlinx.coroutines.Deferred;
import org.reactivestreams.Publisher;
@@ -46,8 +44,9 @@ import org.springframework.util.ReflectionUtils;
* {@code Observable}, and others.
*
* <p>By default, depending on classpath availability, adapters are registered
* for Reactor, RxJava 1, RxJava 2 types, {@link CompletableFuture}, Java 9+
* {@code Flow.Publisher} and Kotlin Coroutines {@code Deferred} and {@code Flow}.
* for Reactor, RxJava 2/3, or RxJava 1 (+ RxJava Reactive Streams bridge),
* {@link CompletableFuture}, Java 9+ {@code Flow.Publisher}, and Kotlin
* Coroutines' {@code Deferred} and {@code Flow}.
*
* @author Rossen Stoyanchev
* @author Sebastien Deleuze
@@ -89,6 +88,11 @@ public class ReactiveAdapterRegistry {
new RxJava2Registrar().registerAdapters(this);
}
// RxJava3
if (ClassUtils.isPresent("io.reactivex.rxjava3.core.Flowable", classLoader)) {
new RxJava3Registrar().registerAdapters(this);
}
// Java 9+ Flow.Publisher
if (ClassUtils.isPresent("java.util.concurrent.Flow.Publisher", classLoader)) {
new ReactorJdkFlowAdapterRegistrar().registerAdapter(this);
@@ -104,9 +108,7 @@ public class ReactiveAdapterRegistry {
/**
* Whether the registry has any adapters which would be the case if any of
* Reactor, RxJava 2, or RxJava 1 (+ RxJava Reactive Streams bridge) are
* present on the classpath.
* Whether the registry has any adapters.
*/
public boolean hasAdapters() {
return !this.adapters.isEmpty();
@@ -254,31 +256,78 @@ public class ReactiveAdapterRegistry {
registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(io.reactivex.Flowable.class, io.reactivex.Flowable::empty),
source -> (io.reactivex.Flowable<?>) source,
Flowable::fromPublisher
io.reactivex.Flowable::fromPublisher
);
registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(io.reactivex.Observable.class, io.reactivex.Observable::empty),
source -> ((io.reactivex.Observable<?>) source).toFlowable(BackpressureStrategy.BUFFER),
source -> io.reactivex.Flowable.fromPublisher(source).toObservable()
source -> ((io.reactivex.Observable<?>) source).toFlowable(io.reactivex.BackpressureStrategy.BUFFER),
source -> io.reactivex.Flowable.fromPublisher(source)
.toObservable()
);
registry.registerReactiveType(
ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.Single.class),
source -> ((io.reactivex.Single<?>) source).toFlowable(),
source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement().toSingle()
source -> io.reactivex.Flowable.fromPublisher(source)
.toObservable().singleElement().toSingle()
);
registry.registerReactiveType(
ReactiveTypeDescriptor.singleOptionalValue(io.reactivex.Maybe.class, io.reactivex.Maybe::empty),
source -> ((io.reactivex.Maybe<?>) source).toFlowable(),
source -> io.reactivex.Flowable.fromPublisher(source).toObservable().singleElement()
source -> io.reactivex.Flowable.fromPublisher(source)
.toObservable().singleElement()
);
registry.registerReactiveType(
ReactiveTypeDescriptor.noValue(io.reactivex.Completable.class, io.reactivex.Completable::complete),
source -> ((io.reactivex.Completable) source).toFlowable(),
source -> io.reactivex.Flowable.fromPublisher(source).toObservable().ignoreElements()
source -> io.reactivex.Flowable.fromPublisher(source)
.toObservable().ignoreElements()
);
}
}
private static class RxJava3Registrar {
void registerAdapters(ReactiveAdapterRegistry registry) {
registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(
io.reactivex.rxjava3.core.Flowable.class,
io.reactivex.rxjava3.core.Flowable::empty),
source -> (io.reactivex.rxjava3.core.Flowable<?>) source,
io.reactivex.rxjava3.core.Flowable::fromPublisher
);
registry.registerReactiveType(
ReactiveTypeDescriptor.multiValue(
io.reactivex.rxjava3.core.Observable.class,
io.reactivex.rxjava3.core.Observable::empty),
source -> ((io.reactivex.rxjava3.core.Observable<?>) source).toFlowable(
io.reactivex.rxjava3.core.BackpressureStrategy.BUFFER),
source -> io.reactivex.rxjava3.core.Flowable.fromPublisher(source)
.toObservable()
);
registry.registerReactiveType(
ReactiveTypeDescriptor.singleRequiredValue(io.reactivex.rxjava3.core.Single.class),
source -> ((io.reactivex.rxjava3.core.Single<?>) source).toFlowable(),
source -> io.reactivex.rxjava3.core.Flowable.fromPublisher(source)
.toObservable().singleElement().toSingle()
);
registry.registerReactiveType(
ReactiveTypeDescriptor.singleOptionalValue(
io.reactivex.rxjava3.core.Maybe.class,
io.reactivex.rxjava3.core.Maybe::empty),
source -> ((io.reactivex.rxjava3.core.Maybe<?>) source).toFlowable(),
source -> io.reactivex.rxjava3.core.Flowable.fromPublisher(source)
.toObservable().singleElement()
);
registry.registerReactiveType(
ReactiveTypeDescriptor.noValue(
io.reactivex.rxjava3.core.Completable.class,
io.reactivex.rxjava3.core.Completable::complete),
source -> ((io.reactivex.rxjava3.core.Completable) source).toFlowable(),
source -> io.reactivex.rxjava3.core.Flowable.fromPublisher(source)
.toObservable().ignoreElements()
);
}
}
private static class ReactorJdkFlowAdapterRegistrar {

6
spring-core/src/test/java/org/springframework/core/ConventionsTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2020 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.
@@ -23,8 +23,8 @@ import java.util.Collections;
import java.util.List;
import java.util.Set;
import io.reactivex.Observable;
import io.reactivex.Single;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Single;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

492
spring-core/src/test/java/org/springframework/core/ReactiveAdapterRegistryTests.java
View File

@@ -21,17 +21,13 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import io.reactivex.Flowable;
import io.reactivex.Maybe;
import kotlinx.coroutines.Deferred;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxProcessor;
import reactor.core.publisher.Mono;
import rx.Completable;
import rx.Observable;
import rx.Single;
import static org.assertj.core.api.Assertions.assertThat;
@@ -45,35 +41,6 @@ class ReactiveAdapterRegistryTests {
private final ReactiveAdapterRegistry registry = ReactiveAdapterRegistry.getSharedInstance();
@Test
void defaultAdapterRegistrations() {
// Reactor
assertThat(getAdapter(Mono.class)).isNotNull();
assertThat(getAdapter(Flux.class)).isNotNull();
// Publisher
assertThat(getAdapter(Publisher.class)).isNotNull();
// Completable
assertThat(getAdapter(CompletableFuture.class)).isNotNull();
// RxJava 1
assertThat(getAdapter(Observable.class)).isNotNull();
assertThat(getAdapter(Single.class)).isNotNull();
assertThat(getAdapter(Completable.class)).isNotNull();
// RxJava 2
assertThat(getAdapter(Flowable.class)).isNotNull();
assertThat(getAdapter(io.reactivex.Observable.class)).isNotNull();
assertThat(getAdapter(io.reactivex.Single.class)).isNotNull();
assertThat(getAdapter(Maybe.class)).isNotNull();
assertThat(getAdapter(io.reactivex.Completable.class)).isNotNull();
// Coroutines
assertThat(getAdapter(Deferred.class)).isNotNull();
}
@Test
void getAdapterForReactiveSubType() {
@@ -93,192 +60,301 @@ class ReactiveAdapterRegistryTests {
assertThat(adapter3).isNotSameAs(adapter1);
}
@Test
void publisherToFlux() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flowable.fromIterable(sequence);
Object target = getAdapter(Flux.class).fromPublisher(source);
boolean condition = target instanceof Flux;
assertThat(condition).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
@Nested
class Reactor {
@Test
void defaultAdapterRegistrations() {
// Reactor
assertThat(getAdapter(Mono.class)).isNotNull();
assertThat(getAdapter(Flux.class)).isNotNull();
// Publisher
assertThat(getAdapter(Publisher.class)).isNotNull();
// Completable
assertThat(getAdapter(CompletableFuture.class)).isNotNull();
}
@Test
void toFlux() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = io.reactivex.rxjava3.core.Flowable.fromIterable(sequence);
Object target = getAdapter(Flux.class).fromPublisher(source);
assertThat(target instanceof Flux).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
}
@Test
void toMono() {
Publisher<Integer> source = io.reactivex.rxjava3.core.Flowable.fromArray(1, 2, 3);
Object target = getAdapter(Mono.class).fromPublisher(source);
assertThat(target instanceof Mono).isTrue();
assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
}
@Test
void toCompletableFuture() throws Exception {
Publisher<Integer> source = Flux.fromArray(new Integer[] {1, 2, 3});
Object target = getAdapter(CompletableFuture.class).fromPublisher(source);
assertThat(target instanceof CompletableFuture).isTrue();
assertThat(((CompletableFuture<Integer>) target).get()).isEqualTo(Integer.valueOf(1));
}
@Test
void fromCompletableFuture() {
CompletableFuture<Integer> future = new CompletableFuture<>();
future.complete(1);
Object target = getAdapter(CompletableFuture.class).toPublisher(future);
assertThat(target instanceof Mono).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
}
}
// TODO: publisherToMono/CompletableFuture vs Single (ISE on multiple elements)?
@Nested
class RxJava1 {
@Test
void publisherToMono() {
Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
Object target = getAdapter(Mono.class).fromPublisher(source);
boolean condition = target instanceof Mono;
assertThat(condition).isTrue();
assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
@Test
void defaultAdapterRegistrations() {
assertThat(getAdapter(rx.Observable.class)).isNotNull();
assertThat(getAdapter(rx.Single.class)).isNotNull();
assertThat(getAdapter(rx.Completable.class)).isNotNull();
}
@Test
void toObservable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flux.fromIterable(sequence);
Object target = getAdapter(rx.Observable.class).fromPublisher(source);
assertThat(target instanceof rx.Observable).isTrue();
assertThat(((rx.Observable<?>) target).toList().toBlocking().first()).isEqualTo(sequence);
}
@Test
void toSingle() {
Publisher<Integer> source = Flux.fromArray(new Integer[] {1});
Object target = getAdapter(rx.Single.class).fromPublisher(source);
assertThat(target instanceof rx.Single).isTrue();
assertThat(((rx.Single<Integer>) target).toBlocking().value()).isEqualTo(Integer.valueOf(1));
}
@Test
void toCompletable() {
Publisher<Integer> source = Flux.fromArray(new Integer[] {1, 2, 3});
Object target = getAdapter(rx.Completable.class).fromPublisher(source);
assertThat(target instanceof rx.Completable).isTrue();
assertThat(((rx.Completable) target).get()).isNull();
}
@Test
void fromObservable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = rx.Observable.from(sequence);
Object target = getAdapter(rx.Observable.class).toPublisher(source);
assertThat(target instanceof Flux).as("Expected Flux Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
}
@Test
void fromSingle() {
Object source = rx.Single.just(1);
Object target = getAdapter(rx.Single.class).toPublisher(source);
assertThat(target instanceof Mono).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
}
@Test
void fromCompletable() {
Object source = rx.Completable.complete();
Object target = getAdapter(rx.Completable.class).toPublisher(source);
assertThat(target instanceof Mono).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
((Mono<Void>) target).block(Duration.ofMillis(1000));
}
}
@Test
void publisherToCompletableFuture() throws Exception {
Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
Object target = getAdapter(CompletableFuture.class).fromPublisher(source);
boolean condition = target instanceof CompletableFuture;
assertThat(condition).isTrue();
assertThat(((CompletableFuture<Integer>) target).get()).isEqualTo(Integer.valueOf(1));
@Nested
class RxJava2 {
@Test
void defaultAdapterRegistrations() {
// RxJava 2
assertThat(getAdapter(io.reactivex.Flowable.class)).isNotNull();
assertThat(getAdapter(io.reactivex.Observable.class)).isNotNull();
assertThat(getAdapter(io.reactivex.Single.class)).isNotNull();
assertThat(getAdapter(io.reactivex.Maybe.class)).isNotNull();
assertThat(getAdapter(io.reactivex.Completable.class)).isNotNull();
}
@Test
void toFlowable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flux.fromIterable(sequence);
Object target = getAdapter(io.reactivex.Flowable.class).fromPublisher(source);
assertThat(target instanceof io.reactivex.Flowable).isTrue();
assertThat(((io.reactivex.Flowable<?>) target).toList().blockingGet()).isEqualTo(sequence);
}
@Test
void toObservable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flux.fromIterable(sequence);
Object target = getAdapter(io.reactivex.Observable.class).fromPublisher(source);
assertThat(target instanceof io.reactivex.Observable).isTrue();
assertThat(((io.reactivex.Observable<?>) target).toList().blockingGet()).isEqualTo(sequence);
}
@Test
void toSingle() {
Publisher<Integer> source = Flux.fromArray(new Integer[] {1});
Object target = getAdapter(io.reactivex.Single.class).fromPublisher(source);
assertThat(target instanceof io.reactivex.Single).isTrue();
assertThat(((io.reactivex.Single<Integer>) target).blockingGet()).isEqualTo(Integer.valueOf(1));
}
@Test
void toCompletable() {
Publisher<Integer> source = Flux.fromArray(new Integer[] {1, 2, 3});
Object target = getAdapter(io.reactivex.Completable.class).fromPublisher(source);
assertThat(target instanceof io.reactivex.Completable).isTrue();
((io.reactivex.Completable) target).blockingAwait();
}
@Test
void fromFlowable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = io.reactivex.Flowable.fromIterable(sequence);
Object target = getAdapter(io.reactivex.Flowable.class).toPublisher(source);
assertThat(target instanceof Flux).as("Expected Flux Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
}
@Test
void fromObservable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = io.reactivex.Observable.fromIterable(sequence);
Object target = getAdapter(io.reactivex.Observable.class).toPublisher(source);
assertThat(target instanceof Flux).as("Expected Flux Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
}
@Test
void fromSingle() {
Object source = io.reactivex.Single.just(1);
Object target = getAdapter(io.reactivex.Single.class).toPublisher(source);
assertThat(target instanceof Mono).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
}
@Test
void fromCompletable() {
Object source = io.reactivex.Completable.complete();
Object target = getAdapter(io.reactivex.Completable.class).toPublisher(source);
assertThat(target instanceof Mono).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
((Mono<Void>) target).block(Duration.ofMillis(1000));
}
}
@Test
void publisherToRxObservable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flowable.fromIterable(sequence);
Object target = getAdapter(rx.Observable.class).fromPublisher(source);
boolean condition = target instanceof Observable;
assertThat(condition).isTrue();
assertThat(((Observable<?>) target).toList().toBlocking().first()).isEqualTo(sequence);
@Nested
class RxJava3 {
@Test
void defaultAdapterRegistrations() {
// RxJava 3
assertThat(getAdapter(io.reactivex.rxjava3.core.Flowable.class)).isNotNull();
assertThat(getAdapter(io.reactivex.rxjava3.core.Observable.class)).isNotNull();
assertThat(getAdapter(io.reactivex.rxjava3.core.Single.class)).isNotNull();
assertThat(getAdapter(io.reactivex.rxjava3.core.Maybe.class)).isNotNull();
assertThat(getAdapter(io.reactivex.rxjava3.core.Completable.class)).isNotNull();
}
@Test
void toFlowable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flux.fromIterable(sequence);
Object target = getAdapter(io.reactivex.rxjava3.core.Flowable.class).fromPublisher(source);
assertThat(target instanceof io.reactivex.rxjava3.core.Flowable).isTrue();
assertThat(((io.reactivex.rxjava3.core.Flowable<?>) target).toList().blockingGet()).isEqualTo(sequence);
}
@Test
void toObservable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flux.fromIterable(sequence);
Object target = getAdapter(io.reactivex.rxjava3.core.Observable.class).fromPublisher(source);
assertThat(target instanceof io.reactivex.rxjava3.core.Observable).isTrue();
assertThat(((io.reactivex.rxjava3.core.Observable<?>) target).toList().blockingGet()).isEqualTo(sequence);
}
@Test
void toSingle() {
Publisher<Integer> source = Flux.fromArray(new Integer[] {1});
Object target = getAdapter(io.reactivex.rxjava3.core.Single.class).fromPublisher(source);
assertThat(target instanceof io.reactivex.rxjava3.core.Single).isTrue();
assertThat(((io.reactivex.rxjava3.core.Single<Integer>) target).blockingGet()).isEqualTo(Integer.valueOf(1));
}
@Test
void toCompletable() {
Publisher<Integer> source = Flux.fromArray(new Integer[] {1, 2, 3});
Object target = getAdapter(io.reactivex.rxjava3.core.Completable.class).fromPublisher(source);
assertThat(target instanceof io.reactivex.rxjava3.core.Completable).isTrue();
((io.reactivex.rxjava3.core.Completable) target).blockingAwait();
}
@Test
void fromFlowable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = io.reactivex.rxjava3.core.Flowable.fromIterable(sequence);
Object target = getAdapter(io.reactivex.rxjava3.core.Flowable.class).toPublisher(source);
assertThat(target instanceof Flux).as("Expected Flux Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
}
@Test
void fromObservable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = io.reactivex.rxjava3.core.Observable.fromIterable(sequence);
Object target = getAdapter(io.reactivex.rxjava3.core.Observable.class).toPublisher(source);
assertThat(target instanceof Flux).as("Expected Flux Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
}
@Test
void fromSingle() {
Object source = io.reactivex.rxjava3.core.Single.just(1);
Object target = getAdapter(io.reactivex.rxjava3.core.Single.class).toPublisher(source);
assertThat(target instanceof Mono).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
}
@Test
void fromCompletable() {
Object source = io.reactivex.rxjava3.core.Completable.complete();
Object target = getAdapter(io.reactivex.rxjava3.core.Completable.class).toPublisher(source);
assertThat(target instanceof Mono).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
((Mono<Void>) target).block(Duration.ofMillis(1000));
}
}
@Test
void publisherToRxSingle() {
Publisher<Integer> source = Flowable.fromArray(1);
Object target = getAdapter(rx.Single.class).fromPublisher(source);
boolean condition = target instanceof Single;
assertThat(condition).isTrue();
assertThat(((Single<Integer>) target).toBlocking().value()).isEqualTo(Integer.valueOf(1));
}
@Nested
class Kotlin {
@Test
void publisherToRxCompletable() {
Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
Object target = getAdapter(rx.Completable.class).fromPublisher(source);
boolean condition = target instanceof Completable;
assertThat(condition).isTrue();
assertThat(((Completable) target).get()).isNull();
}
@Test
void defaultAdapterRegistrations() {
@Test
void publisherToReactivexFlowable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flux.fromIterable(sequence);
Object target = getAdapter(io.reactivex.Flowable.class).fromPublisher(source);
boolean condition = target instanceof Flowable;
assertThat(condition).isTrue();
assertThat(((Flowable<?>) target).toList().blockingGet()).isEqualTo(sequence);
}
// Coroutines
assertThat(getAdapter(Deferred.class)).isNotNull();
}
@Test
void publisherToReactivexObservable() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Publisher<Integer> source = Flowable.fromIterable(sequence);
Object target = getAdapter(io.reactivex.Observable.class).fromPublisher(source);
boolean condition = target instanceof io.reactivex.Observable;
assertThat(condition).isTrue();
assertThat(((io.reactivex.Observable<?>) target).toList().blockingGet()).isEqualTo(sequence);
}
@Test
void publisherToReactivexSingle() {
Publisher<Integer> source = Flowable.fromArray(1);
Object target = getAdapter(io.reactivex.Single.class).fromPublisher(source);
boolean condition = target instanceof io.reactivex.Single;
assertThat(condition).isTrue();
assertThat(((io.reactivex.Single<Integer>) target).blockingGet()).isEqualTo(Integer.valueOf(1));
}
@Test
void publisherToReactivexCompletable() {
Publisher<Integer> source = Flowable.fromArray(1, 2, 3);
Object target = getAdapter(io.reactivex.Completable.class).fromPublisher(source);
boolean condition = target instanceof io.reactivex.Completable;
assertThat(condition).isTrue();
((io.reactivex.Completable) target).blockingAwait();
}
@Test
void rxObservableToPublisher() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = rx.Observable.from(sequence);
Object target = getAdapter(rx.Observable.class).toPublisher(source);
boolean condition = target instanceof Flux;
assertThat(condition).as("Expected Flux Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
}
@Test
void rxSingleToPublisher() {
Object source = rx.Single.just(1);
Object target = getAdapter(rx.Single.class).toPublisher(source);
boolean condition = target instanceof Mono;
assertThat(condition).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
}
@Test
void rxCompletableToPublisher() {
Object source = rx.Completable.complete();
Object target = getAdapter(rx.Completable.class).toPublisher(source);
boolean condition = target instanceof Mono;
assertThat(condition).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
((Mono<Void>) target).block(Duration.ofMillis(1000));
}
@Test
void reactivexFlowableToPublisher() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = io.reactivex.Flowable.fromIterable(sequence);
Object target = getAdapter(io.reactivex.Flowable.class).toPublisher(source);
boolean condition = target instanceof Flux;
assertThat(condition).as("Expected Flux Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
}
@Test
void reactivexObservableToPublisher() {
List<Integer> sequence = Arrays.asList(1, 2, 3);
Object source = io.reactivex.Observable.fromIterable(sequence);
Object target = getAdapter(io.reactivex.Observable.class).toPublisher(source);
boolean condition = target instanceof Flux;
assertThat(condition).as("Expected Flux Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Flux<Integer>) target).collectList().block(Duration.ofMillis(1000))).isEqualTo(sequence);
}
@Test
void reactivexSingleToPublisher() {
Object source = io.reactivex.Single.just(1);
Object target = getAdapter(io.reactivex.Single.class).toPublisher(source);
boolean condition = target instanceof Mono;
assertThat(condition).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
}
@Test
void reactivexCompletableToPublisher() {
Object source = io.reactivex.Completable.complete();
Object target = getAdapter(io.reactivex.Completable.class).toPublisher(source);
boolean condition = target instanceof Mono;
assertThat(condition).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
((Mono<Void>) target).block(Duration.ofMillis(1000));
}
@Test
void completableFutureToPublisher() {
CompletableFuture<Integer> future = new CompletableFuture<>();
future.complete(1);
Object target = getAdapter(CompletableFuture.class).toPublisher(future);
boolean condition = target instanceof Mono;
assertThat(condition).as("Expected Mono Publisher: " + target.getClass().getName()).isTrue();
assertThat(((Mono<Integer>) target).block(Duration.ofMillis(1000))).isEqualTo(Integer.valueOf(1));
}
@Test
void deferred() {
assertThat(getAdapter(CompletableFuture.class).getDescriptor().isDeferred()).isEqualTo(false);
assertThat(getAdapter(Mono.class).getDescriptor().isDeferred()).isEqualTo(true);
assertThat(getAdapter(Flux.class).getDescriptor().isDeferred()).isEqualTo(true);
assertThat(getAdapter(io.reactivex.Completable.class).getDescriptor().isDeferred()).isEqualTo(true);
assertThat(getAdapter(io.reactivex.Single.class).getDescriptor().isDeferred()).isEqualTo(true);
assertThat(getAdapter(io.reactivex.Flowable.class).getDescriptor().isDeferred()).isEqualTo(true);
assertThat(getAdapter(io.reactivex.Observable.class).getDescriptor().isDeferred()).isEqualTo(true);
assertThat(getAdapter(Deferred.class).getDescriptor().isDeferred()).isEqualTo(true);
assertThat(getAdapter(kotlinx.coroutines.flow.Flow.class).getDescriptor().isDeferred()).isEqualTo(true);
@Test
void deferred() {
assertThat(getAdapter(CompletableFuture.class).getDescriptor().isDeferred()).isEqualTo(false);
assertThat(getAdapter(Deferred.class).getDescriptor().isDeferred()).isEqualTo(true);
assertThat(getAdapter(kotlinx.coroutines.flow.Flow.class).getDescriptor().isDeferred()).isEqualTo(true);
}
}
private ReactiveAdapter getAdapter(Class<?> reactiveType) {