-
Notifications
You must be signed in to change notification settings - Fork 3k
/
debounceTime.ts
137 lines (125 loc) · 5 KB
/
debounceTime.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
import { Operator } from '../Operator';
import { Observable } from '../Observable';
import { Subscriber } from '../Subscriber';
import { Subscription } from '../Subscription';
import { async } from '../scheduler/async';
import { MonoTypeOperatorFunction, SchedulerLike, TeardownLogic } from '../types';
/**
* Emits a notification from the source Observable only after a particular time span
* has passed without another source emission.
*
* <span class="informal">It's like {@link delay}, but passes only the most
* recent notification from each burst of emissions.</span>
*
* ![](debounceTime.png)
*
* `debounceTime` delays notifications emitted by the source Observable, but drops
* previous pending delayed emissions if a new notification arrives on the source
* Observable. This operator keeps track of the most recent notification from the
* source Observable, and emits that only when `dueTime` has passed
* without any other notification appearing on the source Observable. If a new value
* appears before `dueTime` silence occurs, the previous notification will be dropped
* and will not be emitted and a new `dueTime` is scheduled.
* If the completing event happens during `dueTime` the last cached notification
* is emitted before the completion event is forwarded to the output observable.
* If the error event happens during `dueTime` or after it only the error event is
* forwarded to the output observable. The cache notification is not emitted in this case.
*
* This is a rate-limiting operator, because it is impossible for more than one
* notification to be emitted in any time window of duration `dueTime`, but it is also
* a delay-like operator since output emissions do not occur at the same time as
* they did on the source Observable. Optionally takes a {@link SchedulerLike} for
* managing timers.
*
* ## Example
* Emit the most recent click after a burst of clicks
* ```ts
* import { fromEvent } from 'rxjs';
* import { debounceTime } from 'rxjs/operators';
*
* const clicks = fromEvent(document, 'click');
* const result = clicks.pipe(debounceTime(1000));
* result.subscribe(x => console.log(x));
* ```
*
* @see {@link audit}
* @see {@link auditTime}
* @see {@link debounce}
* @see {@link debounceTime}
* @see {@link sample}
* @see {@link sampleTime}
* @see {@link throttle}
* @see {@link throttleTime}
*
* @param {number} dueTime The timeout duration in milliseconds (or the time
* unit determined internally by the optional `scheduler`) for the window of
* time required to wait for emission silence before emitting the most recent
* source value.
* @param {SchedulerLike} [scheduler=async] The {@link SchedulerLike} to use for
* managing the timers that handle the timeout for each value.
* @return {Observable} An Observable that delays the emissions of the source
* Observable by the specified `dueTime`, and may drop some values if they occur
* too frequently.
* @method debounceTime
* @owner Observable
*/
export function debounceTime<T>(dueTime: number, scheduler: SchedulerLike = async): MonoTypeOperatorFunction<T> {
return (source: Observable<T>) => source.lift(new DebounceTimeOperator(dueTime, scheduler));
}
class DebounceTimeOperator<T> implements Operator<T, T> {
constructor(private dueTime: number, private scheduler: SchedulerLike) {
}
call(subscriber: Subscriber<T>, source: any): TeardownLogic {
return source.subscribe(new DebounceTimeSubscriber(subscriber, this.dueTime, this.scheduler));
}
}
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class DebounceTimeSubscriber<T> extends Subscriber<T> {
private debouncedSubscription: Subscription = null;
private lastValue: T = null;
private hasValue: boolean = false;
constructor(destination: Subscriber<T>,
private dueTime: number,
private scheduler: SchedulerLike) {
super(destination);
}
protected _next(value: T) {
this.clearDebounce();
this.lastValue = value;
this.hasValue = true;
this.add(this.debouncedSubscription = this.scheduler.schedule(dispatchNext, this.dueTime, this));
}
protected _complete() {
this.debouncedNext();
this.destination.complete();
}
debouncedNext(): void {
this.clearDebounce();
if (this.hasValue) {
const { lastValue } = this;
// This must be done *before* passing the value
// along to the destination because it's possible for
// the value to synchronously re-enter this operator
// recursively when scheduled with things like
// VirtualScheduler/TestScheduler.
this.lastValue = null;
this.hasValue = false;
this.destination.next(lastValue);
}
}
private clearDebounce(): void {
const debouncedSubscription = this.debouncedSubscription;
if (debouncedSubscription !== null) {
this.remove(debouncedSubscription);
debouncedSubscription.unsubscribe();
this.debouncedSubscription = null;
}
}
}
function dispatchNext(subscriber: DebounceTimeSubscriber<any>) {
subscriber.debouncedNext();
}