1import { Operator } from '../Operator';
2import { Subscriber } from '../Subscriber';
3import { Observable } from '../Observable';
4import { Subscription } from '../Subscription';
5import { MonoTypeOperatorFunction, SubscribableOrPromise, TeardownLogic } from '../types';
6
7import { OuterSubscriber } from '../OuterSubscriber';
8import { subscribeToResult } from '../util/subscribeToResult';
9
10/**
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
13 * process.
14 *
15 * <span class="informal">It's like {@link auditTime}, but the silencing
16 * duration is determined by a second Observable.</span>
17 *
18 * ![](audit.png)
19 *
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.
30 *
31 * ## Example
32 *
33 * Emit clicks at a rate of at most one click per second
34 * ```ts
35 * import { fromEvent, interval } from 'rxjs';
36 * import { audit } from 'rxjs/operators'
37 *
38 * const clicks = fromEvent(document, 'click');
39 * const result = clicks.pipe(audit(ev => interval(1000)));
40 * result.subscribe(x => console.log(x));
41 * ```
42 * @see {@link auditTime}
43 * @see {@link debounce}
44 * @see {@link delayWhen}
45 * @see {@link sample}
46 * @see {@link throttle}
47 *
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.
53 * @method audit
54 * @owner Observable
55 */
56export function audit<T>(durationSelector: (value: T) => SubscribableOrPromise<any>): MonoTypeOperatorFunction<T> {
57  return function auditOperatorFunction(source: Observable<T>) {
58    return source.lift(new AuditOperator(durationSelector));
59  };
60}
61
62class AuditOperator<T> implements Operator<T, T> {
63  constructor(private durationSelector: (value: T) => SubscribableOrPromise<any>) {
64  }
65
66  call(subscriber: Subscriber<T>, source: any): TeardownLogic {
67    return source.subscribe(new AuditSubscriber<T, T>(subscriber, this.durationSelector));
68  }
69}
70
71/**
72 * We need this JSDoc comment for affecting ESDoc.
73 * @ignore
74 * @extends {Ignored}
75 */
76class AuditSubscriber<T, R> extends OuterSubscriber<T, R> {
77
78  private value: T;
79  private hasValue: boolean = false;
80  private throttled: Subscription;
81
82  constructor(destination: Subscriber<T>,
83              private durationSelector: (value: T) => SubscribableOrPromise<any>) {
84    super(destination);
85  }
86
87  protected _next(value: T): void {
88    this.value = value;
89    this.hasValue = true;
90    if (!this.throttled) {
91      let duration;
92      try {
93        const { durationSelector } = this;
94        duration = durationSelector(value);
95      } catch (err) {
96        return this.destination.error(err);
97      }
98      const innerSubscription = subscribeToResult(this, duration);
99      if (!innerSubscription || innerSubscription.closed) {
100        this.clearThrottle();
101      } else {
102        this.add(this.throttled = innerSubscription);
103      }
104    }
105  }
106
107  clearThrottle() {
108    const { value, hasValue, throttled } = this;
109    if (throttled) {
110      this.remove(throttled);
111      this.throttled = null;
112      throttled.unsubscribe();
113    }
114    if (hasValue) {
115      this.value = null;
116      this.hasValue = false;
117      this.destination.next(value);
118    }
119  }
120
121  notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number): void {
122    this.clearThrottle();
123  }
124
125  notifyComplete(): void {
126    this.clearThrottle();
127  }
128}
129