Add onDropped callback for throttleWithTimeout - ReactiveX#7458

Author: Desislav-Petrov

src/main/java/io/reactivex/rxjava3/core/Flowable.java

@@ -8930,7 +8930,57 @@ public final Flowable<T> debounce(long timeout, @NonNull TimeUnit unit) {
     public final Flowable<T> debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
         Objects.requireNonNull(unit, "unit is null");
         Objects.requireNonNull(scheduler, "scheduler is null");
-        return RxJavaPlugins.onAssembly(new FlowableDebounceTimed<>(this, timeout, unit, scheduler));
+        return RxJavaPlugins.onAssembly(new FlowableDebounceTimed<>(this, timeout, unit, scheduler, null));
+    }
+
+    /**
+     * Returns a {@code Flowable} that mirrors the current {@code Flowable}, except that it drops items emitted by the
+     * current {@code Flowable} that are followed by newer items before a timeout value expires on a specified
+     * {@link Scheduler}. The timer resets on each emission.
+     * <p>
+     * <em>Note:</em> If items keep being emitted by the current {@code Flowable} faster than the timeout then no items
+     * will be emitted by the resulting {@code Flowable}.
+     * <p>
+     * <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/debounce.s.v3.png" alt="">
+     * <p>
+     * Delivery of the item after the grace period happens on the given {@code Scheduler}'s
+     * {@code Worker} which if takes too long, a newer item may arrive from the upstream, causing the
+     * {@code Worker}'s task to get disposed, which may also interrupt any downstream blocking operation
+     * (yielding an {@code InterruptedException}). It is recommended processing items
+     * that may take long time to be moved to another thread via {@link #observeOn} applied after
+     * {@code debounce} itself.
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>This operator does not support backpressure as it uses time to control data flow.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>You specify which {@code Scheduler} this operator will use.</dd>
+     * </dl>
+     *
+     * @param timeout
+     *            the time each item has to be "the most recent" of those emitted by the current {@code Flowable} to
+     *            ensure that it's not dropped
+     * @param unit
+     *            the unit of time for the specified {@code timeout}
+     * @param scheduler
+     *            the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
+     *            item
+     * @param onDropped
+     *            called with the current entry when it has been replaced by a new one
+     * @return the new {@code Flowable} instance
+     * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
+     * @see <a href="http://reactivex.io/documentation/operators/debounce.html">ReactiveX operators documentation: Debounce</a>
+     * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Backpressure">RxJava wiki: Backpressure</a>
+     * @see #throttleWithTimeout(long, TimeUnit, Scheduler)
+     */
+    @CheckReturnValue
+    @NonNull
+    @BackpressureSupport(BackpressureKind.ERROR)
+    @SchedulerSupport(SchedulerSupport.CUSTOM)
+    public final Flowable<T> debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer<T> onDropped) {
+        Objects.requireNonNull(unit, "unit is null");
+        Objects.requireNonNull(scheduler, "scheduler is null");
+        Objects.requireNonNull(onDropped, "onDropped is null");
+        return RxJavaPlugins.onAssembly(new FlowableDebounceTimed<>(this, timeout, unit, scheduler, onDropped));
     }
 
     /**
@@ -17587,6 +17637,47 @@ public final Flowable<T> throttleWithTimeout(long timeout, @NonNull TimeUnit uni
         return debounce(timeout, unit, scheduler);
     }
 
+    /**
+     * Returns a {@code Flowable} that mirrors the current {@code Flowable}, except that it drops items emitted by the
+     * current {@code Flowable} that are followed by newer items before a timeout value expires on a specified
+     * {@link Scheduler}. The timer resets on each emission (alias to {@link #debounce(long, TimeUnit, Scheduler)}).
+     * <p>
+     * <em>Note:</em> If items keep being emitted by the current {@code Flowable} faster than the timeout then no items
+     * will be emitted by the resulting {@code Flowable}.
+     * <p>
+     * <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/throttleWithTimeout.s.v3.png" alt="">
+     * <dl>
+     *  <dt><b>Backpressure:</b></dt>
+     *  <dd>This operator does not support backpressure as it uses time to control data flow.</dd>
+     *  <dt><b>Scheduler:</b></dt>
+     *  <dd>You specify which {@code Scheduler} this operator will use.</dd>
+     * </dl>
+     *
+     * @param timeout
+     *            the length of the window of time that must pass after the emission of an item from the current
+     *            {@code Flowable} in which it emits no items in order for the item to be emitted by the
+     *            resulting {@code Flowable}
+     * @param unit
+     *            the unit of time for the specified {@code timeout}
+     * @param scheduler
+     *            the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
+     *            item
+     * @param onDropped
+     *            called with the current entry when it has been replaced by a new one
+     * @return the new {@code Flowable} instance
+     * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
+     * @see <a href="http://reactivex.io/documentation/operators/debounce.html">ReactiveX operators documentation: Debounce</a>
+     * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Backpressure">RxJava wiki: Backpressure</a>
+     * @see #debounce(long, TimeUnit, Scheduler)
+     */
+    @CheckReturnValue
+    @BackpressureSupport(BackpressureKind.ERROR)
+    @SchedulerSupport(SchedulerSupport.CUSTOM)
+    @NonNull
+    public final Flowable<T> throttleWithTimeout(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer<T> onDropped) {
+        return debounce(timeout, unit, scheduler, onDropped);
+    }
+
     /**
      * Returns a {@code Flowable} that emits records of the time interval between consecutive items emitted by the
      * current {@code Flowable}.

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounceTimed.java

@@ -16,6 +16,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.*;
 
+import io.reactivex.rxjava3.exceptions.Exceptions;
+import io.reactivex.rxjava3.functions.Consumer;
 import org.reactivestreams.*;
 
 import io.reactivex.rxjava3.core.*;
@@ -32,19 +34,20 @@ public final class FlowableDebounceTimed<T> extends AbstractFlowableWithUpstream
     final long timeout;
     final TimeUnit unit;
     final Scheduler scheduler;
+    final Consumer<T> onDropped;
 
-    public FlowableDebounceTimed(Flowable<T> source, long timeout, TimeUnit unit, Scheduler scheduler) {
+    public FlowableDebounceTimed(Flowable<T> source, long timeout, TimeUnit unit, Scheduler scheduler, Consumer<T> onDropped) {
         super(source);
         this.timeout = timeout;
         this.unit = unit;
         this.scheduler = scheduler;
+        this.onDropped = onDropped;
     }
 
     @Override
     protected void subscribeActual(Subscriber<? super T> s) {
         source.subscribe(new DebounceTimedSubscriber<>(
-                new SerializedSubscriber<>(s),
-                timeout, unit, scheduler.createWorker()));
+                new SerializedSubscriber<>(s), timeout, unit, scheduler.createWorker(), onDropped));
     }
 
     static final class DebounceTimedSubscriber<T> extends AtomicLong
@@ -55,20 +58,22 @@ static final class DebounceTimedSubscriber<T> extends AtomicLong
         final long timeout;
         final TimeUnit unit;
         final Scheduler.Worker worker;
+        final Consumer<T> onDropped;
 
         Subscription upstream;
 
-        Disposable timer;
+        DebounceEmitter<T> timer;
 
         volatile long index;
 
         boolean done;
 
-        DebounceTimedSubscriber(Subscriber<? super T> actual, long timeout, TimeUnit unit, Worker worker) {
+        DebounceTimedSubscriber(Subscriber<? super T> actual, long timeout, TimeUnit unit, Worker worker, Consumer<T> onDropped) {
             this.downstream = actual;
             this.timeout = timeout;
             this.unit = unit;
             this.worker = worker;
+            this.onDropped = onDropped;
         }
 
         @Override
@@ -93,6 +98,18 @@ public void onNext(T t) {
                 d.dispose();
             }
 
+            if (onDropped != null && timer != null) {
+                try {
+                    onDropped.accept(timer.value);
+                } catch (Throwable ex) {
+                    Exceptions.throwIfFatal(ex);
+                    upstream.cancel();
+                    done = true;
+                    downstream.onError(ex);
+                    worker.dispose();
+                }
+            }
+
             DebounceEmitter<T> de = new DebounceEmitter<>(t, idx, this);
             timer = de;
             d = worker.schedule(de, timeout, unit);
@@ -121,15 +138,13 @@ public void onComplete() {
             }
             done = true;
 
-            Disposable d = timer;
+            DebounceEmitter<T> d = timer;
             if (d != null) {
                 d.dispose();
             }
 
-            @SuppressWarnings("unchecked")
-            DebounceEmitter<T> de = (DebounceEmitter<T>)d;
-            if (de != null) {
-                de.emit();
+            if (d != null) {
+                d.emit();
             }
 
             downstream.onComplete();

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableDebounceTest.java

@@ -51,6 +51,39 @@ public void before() {
         innerScheduler = scheduler.createWorker();
     }
 
+    @Test
+    public void debounceWithOnDroppedCallback() {
+        Flowable<String> source = Flowable.unsafeCreate(new Publisher<String>() {
+            @Override
+            public void subscribe(Subscriber<? super String> subscriber) {
+                subscriber.onSubscribe(new BooleanSubscription());
+                publishNext(subscriber, 100, "one");    // Should be skipped since "two" will arrive before the timeout expires.
+                publishNext(subscriber, 400, "two");    // Should be published since "three" will arrive after the timeout expires.
+                publishNext(subscriber, 900, "three");   // Should be skipped since "four" will arrive before the timout expires.
+                publishNext(subscriber, 999, "four");   // Should be skipped since onComplete will arrive before the timeout expires.
+                publishCompleted(subscriber, 1000);     // Should be published as soon as the timeout expires.
+            }
+        });
+
+        Observer<Object> dropCallbackObserver = TestHelper.mockObserver();
+        Flowable<String> sampled = source.debounce(400, TimeUnit.MILLISECONDS, scheduler, dropCallbackObserver::onNext);
+        sampled.subscribe(Subscriber);
+
+        scheduler.advanceTimeTo(0, TimeUnit.MILLISECONDS);
+        InOrder inOrder = inOrder(Subscriber);
+        InOrder dropCallbackOrder = inOrder(dropCallbackObserver);
+
+        // must go to 800 since it must be 400 after when two is sent, which is at 400
+        scheduler.advanceTimeTo(800, TimeUnit.MILLISECONDS);
+        inOrder.verify(Subscriber, times(1)).onNext("two");
+        dropCallbackOrder.verify(dropCallbackObserver, times(1)).onNext("one");
+        scheduler.advanceTimeTo(1000, TimeUnit.MILLISECONDS);
+        dropCallbackOrder.verify(dropCallbackObserver, times(1)).onNext("three");
+        inOrder.verify(Subscriber, times(1)).onComplete();
+        inOrder.verifyNoMoreInteractions();
+        dropCallbackOrder.verifyNoMoreInteractions();
+    }
+
     @Test
     public void debounceWithCompleted() {
         Flowable<String> source = Flowable.unsafeCreate(new Publisher<String>() {
@@ -530,7 +563,7 @@ public void timedBadRequest() {
     public void timedLateEmit() {
         TestSubscriber<Integer> ts = new TestSubscriber<>();
         DebounceTimedSubscriber<Integer> sub = new DebounceTimedSubscriber<>(
-                ts, 1, TimeUnit.SECONDS, new TestScheduler().createWorker());
+                ts, 1, TimeUnit.SECONDS, new TestScheduler().createWorker(), null);
 
         sub.onSubscribe(new BooleanSubscription());
 

src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java

@@ -149,6 +149,7 @@ public void checkParallelFlowable() {
         // negative time is considered as zero time
         addOverride(new ParamOverride(Flowable.class, 0, ParamMode.ANY, "debounce", Long.TYPE, TimeUnit.class));
         addOverride(new ParamOverride(Flowable.class, 0, ParamMode.ANY, "debounce", Long.TYPE, TimeUnit.class, Scheduler.class));
+        addOverride(new ParamOverride(Flowable.class, 0, ParamMode.ANY, "debounce", Long.TYPE, TimeUnit.class, Scheduler.class, Consumer.class));
 
         // null Action allowed
         addOverride(new ParamOverride(Flowable.class, 1, ParamMode.ANY, "onBackpressureBuffer", Long.TYPE, Action.class, BackpressureOverflowStrategy.class));
@@ -177,6 +178,7 @@ public void checkParallelFlowable() {
         // negative time is considered as zero time
         addOverride(new ParamOverride(Flowable.class, 0, ParamMode.ANY, "throttleWithTimeout", Long.TYPE, TimeUnit.class));
         addOverride(new ParamOverride(Flowable.class, 0, ParamMode.ANY, "throttleWithTimeout", Long.TYPE, TimeUnit.class, Scheduler.class));
+        addOverride(new ParamOverride(Flowable.class, 0, ParamMode.ANY, "throttleWithTimeout", Long.TYPE, TimeUnit.class, Scheduler.class, Consumer.class));
 
         // negative time is considered as zero time
         addOverride(new ParamOverride(Flowable.class, 0, ParamMode.ANY, "take", Long.TYPE, TimeUnit.class));