1 import { Operator } from '../Operator';
2 import { Subscriber } from '../Subscriber';
3 import { Observable } from '../Observable';
4 import { Subscription } from '../Subscription';
5 import { MonoTypeOperatorFunction, SubscribableOrPromise, TeardownLogic } from '../types';
7 import { OuterSubscriber } from '../OuterSubscriber';
8 import { subscribeToResult } from '../util/subscribeToResult';
11 * Ignores source values for a duration determined by another Observable, then
12 * emits the most recent value from the source Observable, then repeats this
15 * <span class="informal">It's like {@link auditTime}, but the silencing
16 * duration is determined by a second Observable.</span>
20 * `audit` is similar to `throttle`, but emits the last value from the silenced
21 * time window, instead of the first value. `audit` emits the most recent value
22 * from the source Observable on the output Observable as soon as its internal
23 * timer becomes disabled, and ignores source values while the timer is enabled.
24 * Initially, the timer is disabled. As soon as the first source value arrives,
25 * the timer is enabled by calling the `durationSelector` function with the
26 * source value, which returns the "duration" Observable. When the duration
27 * Observable emits a value or completes, the timer is disabled, then the most
28 * recent source value is emitted on the output Observable, and this process
29 * repeats for the next source value.
33 * Emit clicks at a rate of at most one click per second
35 * import { fromEvent, interval } from 'rxjs';
36 * import { audit } from 'rxjs/operators'
38 * const clicks = fromEvent(document, 'click');
39 * const result = clicks.pipe(audit(ev => interval(1000)));
40 * result.subscribe(x => console.log(x));
42 * @see {@link auditTime}
43 * @see {@link debounce}
44 * @see {@link delayWhen}
46 * @see {@link throttle}
48 * @param {function(value: T): SubscribableOrPromise} durationSelector A function
49 * that receives a value from the source Observable, for computing the silencing
50 * duration, returned as an Observable or a Promise.
51 * @return {Observable<T>} An Observable that performs rate-limiting of
52 * emissions from the source Observable.
56 export function audit<T>(durationSelector: (value: T) => SubscribableOrPromise<any>): MonoTypeOperatorFunction<T> {
57 return function auditOperatorFunction(source: Observable<T>) {
58 return source.lift(new AuditOperator(durationSelector));
62 class AuditOperator<T> implements Operator<T, T> {
63 constructor(private durationSelector: (value: T) => SubscribableOrPromise<any>) {
66 call(subscriber: Subscriber<T>, source: any): TeardownLogic {
67 return source.subscribe(new AuditSubscriber<T, T>(subscriber, this.durationSelector));
72 * We need this JSDoc comment for affecting ESDoc.
76 class AuditSubscriber<T, R> extends OuterSubscriber<T, R> {
79 private hasValue: boolean = false;
80 private throttled: Subscription;
82 constructor(destination: Subscriber<T>,
83 private durationSelector: (value: T) => SubscribableOrPromise<any>) {
87 protected _next(value: T): void {
90 if (!this.throttled) {
93 const { durationSelector } = this;
94 duration = durationSelector(value);
96 return this.destination.error(err);
98 const innerSubscription = subscribeToResult(this, duration);
99 if (!innerSubscription || innerSubscription.closed) {
100 this.clearThrottle();
102 this.add(this.throttled = innerSubscription);
108 const { value, hasValue, throttled } = this;
110 this.remove(throttled);
111 this.throttled = null;
112 throttled.unsubscribe();
116 this.hasValue = false;
117 this.destination.next(value);
121 notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
122 this.clearThrottle();
125 notifyComplete(): void {
126 this.clearThrottle();