From 5a7fbfc1420a6544f7443a38bec5b8e52b058209 Mon Sep 17 00:00:00 2001 From: des Date: Sat, 18 Mar 2023 16:26:36 +0000 Subject: [PATCH] Add onDropped callback for onBackpressureLatest - #7458 --- .../io/reactivex/rxjava3/core/Flowable.java | 41 +++++++++++- .../FlowableOnBackpressureLatest.java | 26 +++++-- .../FlowableOnBackpressureLatestTest.java | 67 ++++++++++++++++++- 3 files changed, 128 insertions(+), 6 deletions(-) diff --git a/src/main/java/io/reactivex/rxjava3/core/Flowable.java b/src/main/java/io/reactivex/rxjava3/core/Flowable.java index d9cea6954f..d8a99ef824 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Flowable.java +++ b/src/main/java/io/reactivex/rxjava3/core/Flowable.java @@ -12755,7 +12755,46 @@ public final Flowable onBackpressureDrop(@NonNull Consumer onDrop) @SchedulerSupport(SchedulerSupport.NONE) @NonNull public final Flowable onBackpressureLatest() { - return RxJavaPlugins.onAssembly(new FlowableOnBackpressureLatest<>(this)); + return RxJavaPlugins.onAssembly(new FlowableOnBackpressureLatest<>(this, null)); + } + + /** + * Drops all but the latest item emitted by the current {@code Flowable} if the downstream is not ready to receive + * new items (indicated by a lack of {@link Subscription#request(long)} calls from it) and emits this latest + * item when the downstream becomes ready. + *

+ * + *

+ * Its behavior is logically equivalent to {@code blockingLatest()} with the exception that + * the downstream is not blocking while requesting more values. + *

+ * Note that if the current {@code Flowable} does support backpressure, this operator ignores that capability + * and doesn't propagate any backpressure requests from downstream. + *

+ * Note that due to the nature of how backpressure requests are propagated through subscribeOn/observeOn, + * requesting more than 1 from downstream doesn't guarantee a continuous delivery of {@code onNext} events. + *

+ *
Backpressure:
+ *
The operator honors backpressure from downstream and consumes the current {@code Flowable} in an unbounded + * manner (i.e., not applying backpressure to it).
+ *
Scheduler:
+ *
{@code onBackpressureLatest} does not operate by default on a particular {@link Scheduler}.
+ *
+ * + * @param onDropped + * called with the current entry when it has been replaced by a new one + * @throws NullPointerException if {@code onDropped} is {@code null} + * @return the new {@code Flowable} instance + * @since 3.1.7 + */ + @CheckReturnValue + @BackpressureSupport(BackpressureKind.UNBOUNDED_IN) + @SchedulerSupport(SchedulerSupport.NONE) + @NonNull + @Experimental + public final Flowable onBackpressureLatest(@NonNull Consumer onDropped) { + Objects.requireNonNull(onDropped, "onDropped is null"); + return RxJavaPlugins.onAssembly(new FlowableOnBackpressureLatest<>(this, onDropped)); } /** diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatest.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatest.java index 1a98831bd2..155e284e93 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatest.java +++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatest.java @@ -14,30 +14,48 @@ package io.reactivex.rxjava3.internal.operators.flowable; import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.exceptions.Exceptions; +import io.reactivex.rxjava3.functions.Consumer; import org.reactivestreams.Subscriber; public final class FlowableOnBackpressureLatest extends AbstractFlowableWithUpstream { - public FlowableOnBackpressureLatest(Flowable source) { + final Consumer onDropped; + + public FlowableOnBackpressureLatest(Flowable source, Consumer onDropped) { super(source); + this.onDropped = onDropped; } @Override protected void subscribeActual(Subscriber s) { - source.subscribe(new BackpressureLatestSubscriber<>(s)); + source.subscribe(new BackpressureLatestSubscriber<>(s, onDropped)); } static final class BackpressureLatestSubscriber extends AbstractBackpressureThrottlingSubscriber { private static final long serialVersionUID = 163080509307634843L; - BackpressureLatestSubscriber(Subscriber downstream) { + final Consumer onDropped; + + BackpressureLatestSubscriber(Subscriber downstream, + Consumer onDropped) { super(downstream); + this.onDropped = onDropped; } @Override public void onNext(T t) { - current.lazySet(t); + T oldValue = current.getAndSet(t); + if (onDropped != null && oldValue != null) { + try { + onDropped.accept(oldValue); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + upstream.cancel(); + downstream.onError(ex); + } + } drain(); } } diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatestTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatestTest.java index eb4cfaf288..f0d50994ed 100644 --- a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatestTest.java +++ b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatestTest.java @@ -17,6 +17,7 @@ import java.util.concurrent.TimeUnit; import org.junit.*; +import org.mockito.InOrder; import org.reactivestreams.Publisher; import io.reactivex.rxjava3.core.*; @@ -27,6 +28,8 @@ import io.reactivex.rxjava3.subscribers.TestSubscriber; import io.reactivex.rxjava3.testsupport.*; +import static org.mockito.Mockito.inOrder; + public class FlowableOnBackpressureLatestTest extends RxJavaTest { @Test public void simple() { @@ -62,6 +65,68 @@ public void simpleBackpressure() { ts.assertNotComplete(); } + @Test + public void simpleBackpressureWithOnDroppedCallback() { + PublishProcessor source = PublishProcessor.create(); + TestSubscriberEx ts = new TestSubscriberEx<>(0L); + + Observer dropCallbackObserver = TestHelper.mockObserver(); + + source.onBackpressureLatest(dropCallbackObserver::onNext) + .subscribe(ts); + + ts.assertNoValues(); + + source.onNext(1); + source.onNext(2); + source.onNext(3); + + ts.request(1); + + ts.assertValues(3); + + source.onNext(4); + source.onNext(5); + + ts.request(2); + + ts.assertValues(3,5); + + InOrder dropCallbackOrder = inOrder(dropCallbackObserver); + dropCallbackOrder.verify(dropCallbackObserver).onNext(1); + dropCallbackOrder.verify(dropCallbackObserver).onNext(2); + dropCallbackOrder.verify(dropCallbackObserver).onNext(4); + dropCallbackOrder.verifyNoMoreInteractions(); + } + + @Test + public void simpleBackpressureWithOnDroppedCallbackEx() { + PublishProcessor source = PublishProcessor.create(); + TestSubscriberEx ts = new TestSubscriberEx<>(0L); + + source.onBackpressureLatest(e -> { + if (e == 3) { + throw new TestException("forced"); + } + }) + .subscribe(ts); + + ts.assertNoValues(); + + source.onNext(1); + source.onNext(2); + + ts.request(1); + + ts.assertValues(2); + + source.onNext(3); + source.onNext(4); + + ts.assertError(TestException.class); + ts.assertValues(2); + } + @Test public void synchronousDrop() { PublishProcessor source = PublishProcessor.create(); @@ -105,7 +170,7 @@ public void synchronousDrop() { } @Test - public void asynchronousDrop() throws InterruptedException { + public void asynchronousDrop() { TestSubscriberEx ts = new TestSubscriberEx(1L) { final Random rnd = new Random(); @Override