auditTime.ts
3.72 KB
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
import { async } from '../scheduler/async';
import { Operator } from '../Operator';
import { Scheduler } from '../Scheduler';
import { Subscriber } from '../Subscriber';
import { Observable } from '../Observable';
import { Subscription, TeardownLogic } from '../Subscription';
/**
* Ignores source values for `duration` milliseconds, then emits the most recent
* value from the source Observable, then repeats this process.
*
* <span class="informal">When it sees a source values, it ignores that plus
* the next ones for `duration` milliseconds, and then it emits the most recent
* value from the source.</span>
*
* <img src="./img/auditTime.png" width="100%">
*
* `auditTime` is similar to `throttleTime`, but emits the last value from the
* silenced time window, instead of the first value. `auditTime` emits the most
* recent value from the source Observable on the output Observable as soon as
* its internal timer becomes disabled, and ignores source values while the
* timer is enabled. Initially, the timer is disabled. As soon as the first
* source value arrives, the timer is enabled. After `duration` milliseconds (or
* the time unit determined internally by the optional `scheduler`) has passed,
* the timer is disabled, then the most recent source value is emitted on the
* output Observable, and this process repeats for the next source value.
* Optionally takes a {@link Scheduler} for managing timers.
*
* @example <caption>Emit clicks at a rate of at most one click per second</caption>
* var clicks = Rx.Observable.fromEvent(document, 'click');
* var result = clicks.auditTime(1000);
* result.subscribe(x => console.log(x));
*
* @see {@link audit}
* @see {@link debounceTime}
* @see {@link delay}
* @see {@link sampleTime}
* @see {@link throttleTime}
*
* @param {number} duration Time to wait before emitting the most recent source
* value, measured in milliseconds or the time unit determined internally
* by the optional `scheduler`.
* @param {Scheduler} [scheduler=async] The {@link Scheduler} to use for
* managing the timers that handle the rate-limiting behavior.
* @return {Observable<T>} An Observable that performs rate-limiting of
* emissions from the source Observable.
* @method auditTime
* @owner Observable
*/
export function auditTime<T>(this: Observable<T>, duration: number, scheduler: Scheduler = async): Observable<T> {
return this.lift(new AuditTimeOperator(duration, scheduler));
}
class AuditTimeOperator<T> implements Operator<T, T> {
constructor(private duration: number,
private scheduler: Scheduler) {
}
call(subscriber: Subscriber<T>, source: any): TeardownLogic {
return source.subscribe(new AuditTimeSubscriber(subscriber, this.duration, this.scheduler));
}
}
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class AuditTimeSubscriber<T> extends Subscriber<T> {
private value: T;
private hasValue: boolean = false;
private throttled: Subscription;
constructor(destination: Subscriber<T>,
private duration: number,
private scheduler: Scheduler) {
super(destination);
}
protected _next(value: T): void {
this.value = value;
this.hasValue = true;
if (!this.throttled) {
this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.duration, this));
}
}
clearThrottle(): void {
const { value, hasValue, throttled } = this;
if (throttled) {
this.remove(throttled);
this.throttled = null;
throttled.unsubscribe();
}
if (hasValue) {
this.value = null;
this.hasValue = false;
this.destination.next(value);
}
}
}
function dispatchNext<T>(subscriber: AuditTimeSubscriber<T>): void {
subscriber.clearThrottle();
}