.config/coc/extensions/node_modules/coc-prettier/node_modules/rxjs/src/internal/operators/audit.ts

   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';
   6
   7 import { OuterSubscriber } from '../OuterSubscriber';
   8 import { 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  */
  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));
  59   };
  60 }
  61
  62 class 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  */
  76 class 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 }