Skip to content

Commit 28f6250

Browse files
Desislav-PetrovDesislav-Petrov
committed
Add onDropped callback for throttleWithTimeout - #7458 
1 parent 65d0739 commit 28f6250

7 files changed

Lines changed: 387 additions & 38 deletions

File tree

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

Lines changed: 92 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8930,7 +8930,57 @@ public final Flowable<T> debounce(long timeout, @NonNull TimeUnit unit) {
89308930
public final Flowable<T> debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
89318931
Objects.requireNonNull(unit, "unit is null");
89328932
Objects.requireNonNull(scheduler, "scheduler is null");
8933-
return RxJavaPlugins.onAssembly(new FlowableDebounceTimed<>(this, timeout, unit, scheduler));
8933+
return RxJavaPlugins.onAssembly(new FlowableDebounceTimed<>(this, timeout, unit, scheduler, null));
8934+
}
8935+
8936+
/**
8937+
* Returns a {@code Flowable} that mirrors the current {@code Flowable}, except that it drops items emitted by the
8938+
* current {@code Flowable} that are followed by newer items before a timeout value expires on a specified
8939+
* {@link Scheduler}. The timer resets on each emission.
8940+
* <p>
8941+
* <em>Note:</em> If items keep being emitted by the current {@code Flowable} faster than the timeout then no items
8942+
* will be emitted by the resulting {@code Flowable}.
8943+
* <p>
8944+
* <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/debounce.s.v3.png" alt="">
8945+
* <p>
8946+
* Delivery of the item after the grace period happens on the given {@code Scheduler}'s
8947+
* {@code Worker} which if takes too long, a newer item may arrive from the upstream, causing the
8948+
* {@code Worker}'s task to get disposed, which may also interrupt any downstream blocking operation
8949+
* (yielding an {@code InterruptedException}). It is recommended processing items
8950+
* that may take long time to be moved to another thread via {@link #observeOn} applied after
8951+
* {@code debounce} itself.
8952+
* <dl>
8953+
* <dt><b>Backpressure:</b></dt>
8954+
* <dd>This operator does not support backpressure as it uses time to control data flow.</dd>
8955+
* <dt><b>Scheduler:</b></dt>
8956+
* <dd>You specify which {@code Scheduler} this operator will use.</dd>
8957+
* </dl>
8958+
*
8959+
* @param timeout
8960+
* the time each item has to be "the most recent" of those emitted by the current {@code Flowable} to
8961+
* ensure that it's not dropped
8962+
* @param unit
8963+
* the unit of time for the specified {@code timeout}
8964+
* @param scheduler
8965+
* the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
8966+
* item
8967+
* @param onDropped
8968+
* called with the current entry when it has been replaced by a new one
8969+
* @return the new {@code Flowable} instance
8970+
* @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
8971+
* @see <a href="http://reactivex.io/documentation/operators/debounce.html">ReactiveX operators documentation: Debounce</a>
8972+
* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Backpressure">RxJava wiki: Backpressure</a>
8973+
* @see #throttleWithTimeout(long, TimeUnit, Scheduler)
8974+
*/
8975+
@CheckReturnValue
8976+
@NonNull
8977+
@BackpressureSupport(BackpressureKind.ERROR)
8978+
@SchedulerSupport(SchedulerSupport.CUSTOM)
8979+
public final Flowable<T> debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer<T> onDropped) {
8980+
Objects.requireNonNull(unit, "unit is null");
8981+
Objects.requireNonNull(scheduler, "scheduler is null");
8982+
Objects.requireNonNull(onDropped, "onDropped is null");
8983+
return RxJavaPlugins.onAssembly(new FlowableDebounceTimed<>(this, timeout, unit, scheduler, onDropped));
89348984
}
89358985

89368986
/**
@@ -17587,6 +17637,47 @@ public final Flowable<T> throttleWithTimeout(long timeout, @NonNull TimeUnit uni
1758717637
return debounce(timeout, unit, scheduler);
1758817638
}
1758917639

17640+
/**
17641+
* Returns a {@code Flowable} that mirrors the current {@code Flowable}, except that it drops items emitted by the
17642+
* current {@code Flowable} that are followed by newer items before a timeout value expires on a specified
17643+
* {@link Scheduler}. The timer resets on each emission (alias to {@link #debounce(long, TimeUnit, Scheduler)}).
17644+
* <p>
17645+
* <em>Note:</em> If items keep being emitted by the current {@code Flowable} faster than the timeout then no items
17646+
* will be emitted by the resulting {@code Flowable}.
17647+
* <p>
17648+
* <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/throttleWithTimeout.s.v3.png" alt="">
17649+
* <dl>
17650+
* <dt><b>Backpressure:</b></dt>
17651+
* <dd>This operator does not support backpressure as it uses time to control data flow.</dd>
17652+
* <dt><b>Scheduler:</b></dt>
17653+
* <dd>You specify which {@code Scheduler} this operator will use.</dd>
17654+
* </dl>
17655+
*
17656+
* @param timeout
17657+
* the length of the window of time that must pass after the emission of an item from the current
17658+
* {@code Flowable} in which it emits no items in order for the item to be emitted by the
17659+
* resulting {@code Flowable}
17660+
* @param unit
17661+
* the unit of time for the specified {@code timeout}
17662+
* @param scheduler
17663+
* the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
17664+
* item
17665+
* @param onDropped
17666+
* called with the current entry when it has been replaced by a new one
17667+
* @return the new {@code Flowable} instance
17668+
* @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
17669+
* @see <a href="http://reactivex.io/documentation/operators/debounce.html">ReactiveX operators documentation: Debounce</a>
17670+
* @see <a href="https://github.com/ReactiveX/RxJava/wiki/Backpressure">RxJava wiki: Backpressure</a>
17671+
* @see #debounce(long, TimeUnit, Scheduler)
17672+
*/
17673+
@CheckReturnValue
17674+
@BackpressureSupport(BackpressureKind.ERROR)
17675+
@SchedulerSupport(SchedulerSupport.CUSTOM)
17676+
@NonNull
17677+
public final Flowable<T> throttleWithTimeout(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer<T> onDropped) {
17678+
return debounce(timeout, unit, scheduler, onDropped);
17679+
}
17680+
1759017681
/**
1759117682
* Returns a {@code Flowable} that emits records of the time interval between consecutive items emitted by the
1759217683
* current {@code Flowable}.

‎src/main/java/io/reactivex/rxjava3/core/Observable.java‎

Lines changed: 84 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7896,7 +7896,53 @@ public final Observable<T> debounce(long timeout, @NonNull TimeUnit unit) {
78967896
public final Observable<T> debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
78977897
Objects.requireNonNull(unit, "unit is null");
78987898
Objects.requireNonNull(scheduler, "scheduler is null");
7899-
return RxJavaPlugins.onAssembly(new ObservableDebounceTimed<>(this, timeout, unit, scheduler));
7899+
return RxJavaPlugins.onAssembly(new ObservableDebounceTimed<>(this, timeout, unit, scheduler, null));
7900+
}
7901+
7902+
/**
7903+
* Returns an {@code Observable} that mirrors the current {@code Observable}, except that it drops items emitted by the
7904+
* current {@code Observable} that are followed by newer items before a timeout value expires on a specified
7905+
* {@link Scheduler}. The timer resets on each emission.
7906+
* <p>
7907+
* <em>Note:</em> If items keep being emitted by the current {@code Observable} faster than the timeout then no items
7908+
* will be emitted by the resulting {@code Observable}.
7909+
* <p>
7910+
* <img width="640" height="310" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/debounce.s.v3.png" alt="">
7911+
* <p>
7912+
* Delivery of the item after the grace period happens on the given {@code Scheduler}'s
7913+
* {@code Worker} which if takes too long, a newer item may arrive from the upstream, causing the
7914+
* {@code Worker}'s task to get disposed, which may also interrupt any downstream blocking operation
7915+
* (yielding an {@code InterruptedException}). It is recommended processing items
7916+
* that may take long time to be moved to another thread via {@link #observeOn} applied after
7917+
* {@code debounce} itself.
7918+
* <dl>
7919+
* <dt><b>Scheduler:</b></dt>
7920+
* <dd>You specify which {@code Scheduler} this operator will use.</dd>
7921+
* </dl>
7922+
*
7923+
* @param timeout
7924+
* the time each item has to be "the most recent" of those emitted by the current {@code Observable} to
7925+
* ensure that it's not dropped
7926+
* @param unit
7927+
* the unit of time for the specified {@code timeout}
7928+
* @param scheduler
7929+
* the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
7930+
* item
7931+
* @param onDropped
7932+
* called with the current entry when it has been replaced by a new one
7933+
* @return the new {@code Observable} instance
7934+
* @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} } or {@code onDropped} is {@code null}
7935+
* @see <a href="http://reactivex.io/documentation/operators/debounce.html">ReactiveX operators documentation: Debounce</a>
7936+
* @see #throttleWithTimeout(long, TimeUnit, Scheduler)
7937+
*/
7938+
@CheckReturnValue
7939+
@SchedulerSupport(SchedulerSupport.CUSTOM)
7940+
@NonNull
7941+
public final Observable<T> debounce(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer<T> onDropped) {
7942+
Objects.requireNonNull(unit, "unit is null");
7943+
Objects.requireNonNull(scheduler, "scheduler is null");
7944+
Objects.requireNonNull(onDropped, "onDropped is null");
7945+
return RxJavaPlugins.onAssembly(new ObservableDebounceTimed<>(this, timeout, unit, scheduler, onDropped));
79007946
}
79017947

79027948
/**
@@ -14597,6 +14643,43 @@ public final Observable<T> throttleWithTimeout(long timeout, @NonNull TimeUnit u
1459714643
return debounce(timeout, unit, scheduler);
1459814644
}
1459914645

14646+
/**
14647+
* Returns an {@code Observable} that mirrors the current {@code Observable}, except that it drops items emitted by the
14648+
* current {@code Observable} that are followed by newer items before a timeout value expires on a specified
14649+
* {@link Scheduler}. The timer resets on each emission (Alias to {@link #debounce(long, TimeUnit, Scheduler)}).
14650+
* <p>
14651+
* <em>Note:</em> If items keep being emitted by the current {@code Observable} faster than the timeout then no items
14652+
* will be emitted by the resulting {@code Observable}.
14653+
* <p>
14654+
* <img width="640" height="305" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/throttleWithTimeout.s.v3.png" alt="">
14655+
* <dl>
14656+
* <dt><b>Scheduler:</b></dt>
14657+
* <dd>You specify which {@code Scheduler} this operator will use.</dd>
14658+
* </dl>
14659+
*
14660+
* @param timeout
14661+
* the length of the window of time that must pass after the emission of an item from the current
14662+
* {@code Observable}, in which the current {@code Observable} emits no items, in order for the item to be emitted by the
14663+
* resulting {@code Observable}
14664+
* @param unit
14665+
* the unit of time for the specified {@code timeout}
14666+
* @param scheduler
14667+
* the {@code Scheduler} to use internally to manage the timers that handle the timeout for each
14668+
* item
14669+
* @param onDropped
14670+
* called with the current entry when it has been replaced by a new one
14671+
* @return the new {@code Observable} instance
14672+
* @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null} or {@code onDropped} is {@code null}
14673+
* @see <a href="http://reactivex.io/documentation/operators/debounce.html">ReactiveX operators documentation: Debounce</a>
14674+
* @see #debounce(long, TimeUnit, Scheduler)
14675+
*/
14676+
@CheckReturnValue
14677+
@SchedulerSupport(SchedulerSupport.CUSTOM)
14678+
@NonNull
14679+
public final Observable<T> throttleWithTimeout(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Consumer<T> onDropped) {
14680+
return debounce(timeout, unit, scheduler, onDropped);
14681+
}
14682+
1460014683
/**
1460114684
* Returns an {@code Observable} that emits records of the time interval between consecutive items emitted by the
1460214685
* current {@code Observable}.

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

Lines changed: 25 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
import java.util.concurrent.TimeUnit;
1717
import java.util.concurrent.atomic.*;
1818

19+
import io.reactivex.rxjava3.exceptions.Exceptions;
20+
import io.reactivex.rxjava3.functions.Consumer;
1921
import org.reactivestreams.*;
2022

2123
import io.reactivex.rxjava3.core.*;
@@ -32,19 +34,20 @@ public final class FlowableDebounceTimed<T> extends AbstractFlowableWithUpstream
3234
final long timeout;
3335
final TimeUnit unit;
3436
final Scheduler scheduler;
37+
final Consumer<T> onDropped;
3538

36-
public FlowableDebounceTimed(Flowable<T> source, long timeout, TimeUnit unit, Scheduler scheduler) {
39+
public FlowableDebounceTimed(Flowable<T> source, long timeout, TimeUnit unit, Scheduler scheduler, Consumer<T> onDropped) {
3740
super(source);
3841
this.timeout = timeout;
3942
this.unit = unit;
4043
this.scheduler = scheduler;
44+
this.onDropped = onDropped;
4145
}
4246

4347
@Override
4448
protected void subscribeActual(Subscriber<? super T> s) {
4549
source.subscribe(new DebounceTimedSubscriber<>(
46-
new SerializedSubscriber<>(s),
47-
timeout, unit, scheduler.createWorker()));
50+
new SerializedSubscriber<>(s), timeout, unit, scheduler.createWorker(), onDropped));
4851
}
4952

5053
static final class DebounceTimedSubscriber<T> extends AtomicLong
@@ -55,20 +58,22 @@ static final class DebounceTimedSubscriber<T> extends AtomicLong
5558
final long timeout;
5659
final TimeUnit unit;
5760
final Scheduler.Worker worker;
61+
final Consumer<T> onDropped;
5862

5963
Subscription upstream;
6064

61-
Disposable timer;
65+
DebounceEmitter<T> timer;
6266

6367
volatile long index;
6468

6569
boolean done;
6670

67-
DebounceTimedSubscriber(Subscriber<? super T> actual, long timeout, TimeUnit unit, Worker worker) {
71+
DebounceTimedSubscriber(Subscriber<? super T> actual, long timeout, TimeUnit unit, Worker worker, Consumer<T> onDropped) {
6872
this.downstream = actual;
6973
this.timeout = timeout;
7074
this.unit = unit;
7175
this.worker = worker;
76+
this.onDropped = onDropped;
7277
}
7378

7479
@Override
@@ -93,6 +98,18 @@ public void onNext(T t) {
9398
d.dispose();
9499
}
95100

101+
if (onDropped != null && timer != null) {
102+
try {
103+
onDropped.accept(timer.value);
104+
} catch (Throwable ex) {
105+
Exceptions.throwIfFatal(ex);
106+
upstream.cancel();
107+
done = true;
108+
downstream.onError(ex);
109+
worker.dispose();
110+
}
111+
}
112+
96113
DebounceEmitter<T> de = new DebounceEmitter<>(t, idx, this);
97114
timer = de;
98115
d = worker.schedule(de, timeout, unit);
@@ -121,15 +138,13 @@ public void onComplete() {
121138
}
122139
done = true;
123140

124-
Disposable d = timer;
141+
DebounceEmitter<T> d = timer;
125142
if (d != null) {
126143
d.dispose();
127144
}
128145

129-
@SuppressWarnings("unchecked")
130-
DebounceEmitter<T> de = (DebounceEmitter<T>)d;
131-
if (de != null) {
132-
de.emit();
146+
if (d != null) {
147+
d.emit();
133148
}
134149

135150
downstream.onComplete();

‎src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableDebounceTimed.java‎

Lines changed: 24 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919
import io.reactivex.rxjava3.core.*;
2020
import io.reactivex.rxjava3.core.Scheduler.Worker;
2121
import io.reactivex.rxjava3.disposables.Disposable;
22+
import io.reactivex.rxjava3.exceptions.Exceptions;
23+
import io.reactivex.rxjava3.functions.Consumer;
2224
import io.reactivex.rxjava3.internal.disposables.DisposableHelper;
2325
import io.reactivex.rxjava3.observers.SerializedObserver;
2426
import io.reactivex.rxjava3.plugins.RxJavaPlugins;
@@ -27,19 +29,20 @@ public final class ObservableDebounceTimed<T> extends AbstractObservableWithUpst
2729
final long timeout;
2830
final TimeUnit unit;
2931
final Scheduler scheduler;
32+
final Consumer<T> onDropped;
3033

31-
public ObservableDebounceTimed(ObservableSource<T> source, long timeout, TimeUnit unit, Scheduler scheduler) {
34+
public ObservableDebounceTimed(ObservableSource<T> source, long timeout, TimeUnit unit, Scheduler scheduler, Consumer<T> onDropped) {
3235
super(source);
3336
this.timeout = timeout;
3437
this.unit = unit;
3538
this.scheduler = scheduler;
39+
this.onDropped = onDropped;
3640
}
3741

3842
@Override
3943
public void subscribeActual(Observer<? super T> t) {
4044
source.subscribe(new DebounceTimedObserver<>(
41-
new SerializedObserver<>(t),
42-
timeout, unit, scheduler.createWorker()));
45+
new SerializedObserver<>(t), timeout, unit, scheduler.createWorker(), onDropped));
4346
}
4447

4548
static final class DebounceTimedObserver<T>
@@ -48,20 +51,22 @@ static final class DebounceTimedObserver<T>
4851
final long timeout;
4952
final TimeUnit unit;
5053
final Scheduler.Worker worker;
54+
final Consumer<T> onDropped;
5155

5256
Disposable upstream;
5357

54-
Disposable timer;
58+
DebounceEmitter<T> timer;
5559

5660
volatile long index;
5761

5862
boolean done;
5963

60-
DebounceTimedObserver(Observer<? super T> actual, long timeout, TimeUnit unit, Worker worker) {
64+
DebounceTimedObserver(Observer<? super T> actual, long timeout, TimeUnit unit, Worker worker, Consumer<T> onDropped) {
6165
this.downstream = actual;
6266
this.timeout = timeout;
6367
this.unit = unit;
6468
this.worker = worker;
69+
this.onDropped = onDropped;
6570
}
6671

6772
@Override
@@ -85,6 +90,17 @@ public void onNext(T t) {
8590
d.dispose();
8691
}
8792

93+
if (onDropped != null && timer != null) {
94+
try {
95+
onDropped.accept(timer.value);
96+
} catch (Throwable ex) {
97+
Exceptions.throwIfFatal(ex);
98+
upstream.dispose();
99+
downstream.onError(ex);
100+
done = true;
101+
}
102+
}
103+
88104
DebounceEmitter<T> de = new DebounceEmitter<>(t, idx, this);
89105
timer = de;
90106
d = worker.schedule(de, timeout, unit);
@@ -113,15 +129,13 @@ public void onComplete() {
113129
}
114130
done = true;
115131

116-
Disposable d = timer;
132+
DebounceEmitter<T> d = timer;
117133
if (d != null) {
118134
d.dispose();
119135
}
120136

121-
@SuppressWarnings("unchecked")
122-
DebounceEmitter<T> de = (DebounceEmitter<T>)d;
123-
if (de != null) {
124-
de.run();
137+
if (d != null) {
138+
d.run();
125139
}
126140
downstream.onComplete();
127141
worker.dispose();

0 commit comments

Comments
 (0)