From 80299838b3cd3d1ba9d9ba6f19da8aa8dfd4dd20 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Mon, 6 Jul 2020 11:16:07 +0200 Subject: [PATCH] DATACMNS-1763 - Expose ReactiveWrapperConverters.doOnSuccess and doOnError callback hooks. We now provide callback hook operators that allow attaching of a Runnable and Consumer to a reactive wrapper type for participating in terminal signals. --- .../util/ReactiveWrapperConverters.java | 261 +++++++++++++++++- .../ReactiveWrapperConvertersUnitTests.java | 46 +++ 2 files changed, 305 insertions(+), 2 deletions(-) diff --git a/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java b/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java index 5995f1439..d3c68c0f4 100644 --- a/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java +++ b/src/main/java/org/springframework/data/repository/util/ReactiveWrapperConverters.java @@ -18,6 +18,7 @@ package org.springframework.data.repository.util; import io.reactivex.Flowable; import io.reactivex.Maybe; import kotlinx.coroutines.flow.Flow; +import kotlinx.coroutines.flow.FlowKt; import kotlinx.coroutines.reactive.ReactiveFlowKt; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -26,6 +27,7 @@ import rx.Single; import java.util.ArrayList; import java.util.List; +import java.util.function.Consumer; import java.util.function.Function; import javax.annotation.Nonnull; @@ -97,6 +99,10 @@ public abstract class ReactiveWrapperConverters { REACTIVE_WRAPPERS.add(PublisherWrapper.INSTANCE); } + if (ReactiveWrappers.isAvailable(ReactiveLibrary.KOTLIN_COROUTINES)) { + REACTIVE_WRAPPERS.add(FlowWrapper.INSTANCE); + } + registerConvertersIn(GENERIC_CONVERSION_SERVICE); } @@ -184,6 +190,50 @@ public abstract class ReactiveWrapperConverters { .orElseThrow(() -> new IllegalStateException(String.format("Cannot apply converter to %s", reactiveObject))); } + /** + * Apply a {@link Runnable} when the reactive type emits a completion signal. + * + * @param reactiveObject must not be {@literal null}. + * @param onSuccess must not be {@literal null}. + * @return + * @since 2.4 + */ + @SuppressWarnings("unchecked") + public static T doOnSuccess(Object reactiveObject, Runnable onSuccess) { + + Assert.notNull(reactiveObject, "Reactive source object must not be null!"); + Assert.notNull(onSuccess, "onSuccess callback must not be null!"); + + return REACTIVE_WRAPPERS.stream()// + .filter(it -> ClassUtils.isAssignable(it.getWrapperClass(), reactiveObject.getClass()))// + .findFirst()// + .map(it -> (T) it.doOnSuccess(reactiveObject, onSuccess))// + .orElseThrow( + () -> new IllegalStateException(String.format("Cannot apply onSuccess callback to %s", reactiveObject))); + } + + /** + * Apply a {@link Consumer} when the reactive type emits an error signal. + * + * @param reactiveObject must not be {@literal null}. + * @param onError must not be {@literal null}. + * @return + * @since 2.4 + */ + @SuppressWarnings("unchecked") + public static T doOnError(Object reactiveObject, Consumer onError) { + + Assert.notNull(reactiveObject, "Reactive source object must not be null!"); + Assert.notNull(onError, "onError callback must not be null!"); + + return REACTIVE_WRAPPERS.stream()// + .filter(it -> ClassUtils.isAssignable(it.getWrapperClass(), reactiveObject.getClass()))// + .findFirst()// + .map(it -> (T) it.doOnError(reactiveObject, onError))// + .orElseThrow( + () -> new IllegalStateException(String.format("Cannot apply onError callback to %s", reactiveObject))); + } + /** * Return {@literal true} if objects of {@code sourceType} can be converted to the {@code targetType}. * @@ -209,7 +259,7 @@ public abstract class ReactiveWrapperConverters { * @author Mark Paluch * @author Christoph Strobl */ - private static interface ReactiveTypeWrapper { + private interface ReactiveTypeWrapper { /** * @return the wrapper class. @@ -224,6 +274,26 @@ public abstract class ReactiveWrapperConverters { * @return the reactive type applying conversion. */ Object map(Object wrapper, Function function); + + /** + * Apply a {@link Runnable} when the reactive type emits a completion signal. + * + * @param wrapper the reactive type, must not be {@literal null}. + * @param onSuccess the signal callback, must not be {@literal null}. + * @return the reactive type with {@code onSuccess} attached. + * @since 2.4 + */ + Object doOnSuccess(Object wrapper, Runnable onSuccess); + + /** + * Apply a {@link Consumer} when the reactive type emits an error signal. + * + * @param wrapper the reactive type, must not be {@literal null}. + * @param onError the error consumer, must not be {@literal null}. + * @return the reactive type with {@code onError} attached. + * @since 2.4 + */ + Object doOnError(Object wrapper, Consumer onError); } /** @@ -242,6 +312,16 @@ public abstract class ReactiveWrapperConverters { public Mono map(Object wrapper, Function function) { return ((Mono) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((Mono) wrapper).doOnSuccess(o -> onSuccess.run()); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((Mono) wrapper).doOnError(onError); + } } /** @@ -257,7 +337,56 @@ public abstract class ReactiveWrapperConverters { } public Flux map(Object wrapper, Function function) { - return ((Flux) wrapper).map(function::apply); + return ((Flux) wrapper).map(function); + } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((Flux) wrapper).doOnComplete(onSuccess); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((Flux) wrapper).doOnError(onError); + } + } + + /** + * Wrapper for Kotlin's {@link Flow}. + * + * @since 2.4 + */ + private enum FlowWrapper implements ReactiveTypeWrapper> { + + INSTANCE; + + @Override + public Class> getWrapperClass() { + return Flow.class; + } + + public Flow map(Object wrapper, Function function) { + return FlowKt.map((Flow) wrapper, (o, continuation) -> function.apply(o)); + } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return FlowKt.onCompletion((Flow) wrapper, (collector, ex, continuation) -> { + if (ex == null) { + onSuccess.run(); + } + return collector; + }); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return FlowKt.onCompletion((Flow) wrapper, (collector, ex, continuation) -> { + if (ex != null) { + onError.accept(ex); + } + return collector; + }); } } @@ -286,6 +415,34 @@ public abstract class ReactiveWrapperConverters { return FluxWrapper.INSTANCE.map(Flux.from((Publisher) wrapper), function); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + + if (wrapper instanceof Mono) { + return MonoWrapper.INSTANCE.doOnSuccess(wrapper, onSuccess); + } + + if (wrapper instanceof Flux) { + return FluxWrapper.INSTANCE.doOnSuccess(wrapper, onSuccess); + } + + return FluxWrapper.INSTANCE.doOnSuccess(Flux.from((Publisher) wrapper), onSuccess); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + + if (wrapper instanceof Mono) { + return MonoWrapper.INSTANCE.doOnError(wrapper, onError); + } + + if (wrapper instanceof Flux) { + return FluxWrapper.INSTANCE.doOnError(wrapper, onError); + } + + return FluxWrapper.INSTANCE.doOnError(Flux.from((Publisher) wrapper), onError); + } } // ------------------------------------------------------------------------- @@ -308,6 +465,16 @@ public abstract class ReactiveWrapperConverters { public Single map(Object wrapper, Function function) { return ((Single) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((Single) wrapper).doOnSuccess(o -> onSuccess.run()); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((Single) wrapper).doOnError(onError::accept); + } } /** @@ -326,6 +493,16 @@ public abstract class ReactiveWrapperConverters { public Observable map(Object wrapper, Function function) { return ((Observable) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((Observable) wrapper).doOnCompleted(onSuccess::run); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((Observable) wrapper).doOnError(onError::accept); + } } // ------------------------------------------------------------------------- @@ -348,6 +525,16 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.Single map(Object wrapper, Function function) { return ((io.reactivex.Single) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((io.reactivex.Single) wrapper).doOnSuccess(o -> onSuccess.run()); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((io.reactivex.Single) wrapper).doOnError(onError::accept); + } } /** @@ -366,6 +553,16 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.Maybe map(Object wrapper, Function function) { return ((io.reactivex.Maybe) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((io.reactivex.Maybe) wrapper).doOnSuccess(o -> onSuccess.run()); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((io.reactivex.Maybe) wrapper).doOnError(onError::accept); + } } /** @@ -384,6 +581,16 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.Observable map(Object wrapper, Function function) { return ((io.reactivex.Observable) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((io.reactivex.Observable) wrapper).doOnComplete(onSuccess::run); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((io.reactivex.Observable) wrapper).doOnError(onError::accept); + } } /** @@ -402,6 +609,16 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.Flowable map(Object wrapper, Function function) { return ((io.reactivex.Flowable) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((io.reactivex.Flowable) wrapper).doOnComplete(onSuccess::run); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((io.reactivex.Flowable) wrapper).doOnError(onError::accept); + } } // ------------------------------------------------------------------------- @@ -424,6 +641,16 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.rxjava3.core.Single map(Object wrapper, Function function) { return ((io.reactivex.rxjava3.core.Single) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((io.reactivex.rxjava3.core.Single) wrapper).doOnSuccess(o -> onSuccess.run()); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((io.reactivex.rxjava3.core.Single) wrapper).doOnError(onError::accept); + } } /** @@ -442,6 +669,16 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.rxjava3.core.Maybe map(Object wrapper, Function function) { return ((io.reactivex.rxjava3.core.Maybe) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((io.reactivex.rxjava3.core.Maybe) wrapper).doOnSuccess(o -> onSuccess.run()); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((io.reactivex.rxjava3.core.Maybe) wrapper).doOnError(onError::accept); + } } /** @@ -460,6 +697,16 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.rxjava3.core.Observable map(Object wrapper, Function function) { return ((io.reactivex.rxjava3.core.Observable) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((io.reactivex.rxjava3.core.Observable) wrapper).doOnComplete(onSuccess::run); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((io.reactivex.rxjava3.core.Observable) wrapper).doOnError(onError::accept); + } } /** @@ -478,6 +725,16 @@ public abstract class ReactiveWrapperConverters { public io.reactivex.rxjava3.core.Flowable map(Object wrapper, Function function) { return ((io.reactivex.rxjava3.core.Flowable) wrapper).map(function::apply); } + + @Override + public Object doOnSuccess(Object wrapper, Runnable onSuccess) { + return ((io.reactivex.rxjava3.core.Flowable) wrapper).doOnComplete(onSuccess::run); + } + + @Override + public Object doOnError(Object wrapper, Consumer onError) { + return ((io.reactivex.rxjava3.core.Flowable) wrapper).doOnError(onError::accept); + } } // ------------------------------------------------------------------------- diff --git a/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java b/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java index bb64dfd30..b5c2518b7 100644 --- a/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java +++ b/src/test/java/org/springframework/data/repository/util/ReactiveWrapperConvertersUnitTests.java @@ -18,12 +18,18 @@ package org.springframework.data.repository.util; import static org.assertj.core.api.AssertionsForClassTypes.*; import io.reactivex.Maybe; +import kotlinx.coroutines.flow.Flow; +import kotlinx.coroutines.flow.FlowKt; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; import rx.Completable; import rx.Observable; import rx.Single; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + import org.junit.jupiter.api.Test; import org.reactivestreams.Publisher; @@ -71,6 +77,13 @@ class ReactiveWrapperConvertersUnitTests { assertThat(ReactiveWrapperConverters.supports(io.reactivex.rxjava3.core.Completable.class)).isTrue(); } + @Test // DATACMNS-1763 + void shouldSupportKotlinFlow() { + + assertThat(ReactiveWrapperConverters.supports(Flow.class)).isTrue(); + assertThat(ReactiveWrapperConverters.supports(io.reactivex.rxjava3.core.Completable.class)).isTrue(); + } + @Test // DATACMNS-836 void toWrapperShouldCastMonoToMono() { @@ -216,6 +229,30 @@ class ReactiveWrapperConvertersUnitTests { assertThat(map.block()).isEqualTo(1L); } + @Test // DATACMNS-1763 + void shouldDoOnSuccess() { + + Mono foo = Mono.just("foo"); + AtomicBoolean success = new AtomicBoolean(); + Mono doOn = ReactiveWrapperConverters.doOnSuccess(foo, () -> success.set(true)); + + doOn.as(StepVerifier::create).expectNext("foo").verifyComplete(); + + assertThat(success.get()).isTrue(); + } + + @Test // DATACMNS-1763 + void shouldDoOnError() { + + Mono foo = Mono.error(new IllegalStateException()); + AtomicReference error = new AtomicReference<>(); + Mono doOn = ReactiveWrapperConverters.doOnError(foo, error::set); + + doOn.as(StepVerifier::create).verifyError(); + + assertThat(error.get()).isInstanceOf(IllegalStateException.class); + } + @Test // DATACMNS-836 void shouldMapFlux() { @@ -303,4 +340,13 @@ class ReactiveWrapperConvertersUnitTests { io.reactivex.rxjava3.core.Flowable map = ReactiveWrapperConverters.map(foo, source -> 1L); assertThat(map.blockingFirst()).isEqualTo(1L); } + + @Test // DATACMNS-1763 + @SuppressWarnings("deprecation") + void shouldMapKotlinFlow() { + + Flow flow = FlowKt.asFlow(new String[] { "foo" }); + Flow map = ReactiveWrapperConverters.map(flow, source -> 1L); + StepVerifier.create(kotlinx.coroutines.reactive.FlowKt.asPublisher(map)).expectNext(1L).verifyComplete(); + } }