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