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

   1 import { Operator } from '../Operator';
   2 import { Subscriber } from '../Subscriber';
   3 import { Observable } from '../Observable';
   4 import { Subject } from '../Subject';
   5 import { Subscription } from '../Subscription';
   6
   7 import { OuterSubscriber } from '../OuterSubscriber';
   8 import { InnerSubscriber } from '../InnerSubscriber';
   9 import { subscribeToResult } from '../util/subscribeToResult';
  10
  11 import { MonoTypeOperatorFunction, TeardownLogic } from '../types';
  12
  13 /**
  14  * Returns an Observable that mirrors the source Observable with the exception of a `complete`. If the source
  15  * Observable calls `complete`, this method will emit to the Observable returned from `notifier`. If that Observable
  16  * calls `complete` or `error`, then this method will call `complete` or `error` on the child subscription. Otherwise
  17  * this method will resubscribe to the source Observable.
  18  *
  19  * ![](repeatWhen.png)
  20  *
  21  * ## Example
  22  * Repeat a message stream on click
  23  * ```ts
  24  * import { of, fromEvent } from 'rxjs';
  25  * import { repeatWhen } from 'rxjs/operators';
  26  *
  27  * const source = of('Repeat message');
  28  * const documentClick$ = fromEvent(document, 'click');
  29  *
  30  * source.pipe(repeatWhen(() => documentClick$)
  31  * ).subscribe(data => console.log(data))
  32  * ```
  33  * @see {@link repeat}
  34  * @see {@link retry}
  35  * @see {@link retryWhen}
  36  *
  37  * @param {function(notifications: Observable): Observable} notifier - Receives an Observable of notifications with
  38  * which a user can `complete` or `error`, aborting the repetition.
  39  * @return {Observable} The source Observable modified with repeat logic.
  40  * @method repeatWhen
  41  * @owner Observable
  42  */
  43 export function repeatWhen<T>(notifier: (notifications: Observable<any>) => Observable<any>): MonoTypeOperatorFunction<T> {
  44   return (source: Observable<T>) => source.lift(new RepeatWhenOperator(notifier));
  45 }
  46
  47 class RepeatWhenOperator<T> implements Operator<T, T> {
  48   constructor(protected notifier: (notifications: Observable<any>) => Observable<any>) {
  49   }
  50
  51   call(subscriber: Subscriber<T>, source: any): TeardownLogic {
  52     return source.subscribe(new RepeatWhenSubscriber(subscriber, this.notifier, source));
  53   }
  54 }
  55
  56 /**
  57  * We need this JSDoc comment for affecting ESDoc.
  58  * @ignore
  59  * @extends {Ignored}
  60  */
  61 class RepeatWhenSubscriber<T, R> extends OuterSubscriber<T, R> {
  62
  63   private notifications: Subject<any>;
  64   private retries: Observable<any>;
  65   private retriesSubscription: Subscription;
  66   private sourceIsBeingSubscribedTo: boolean = true;
  67
  68   constructor(destination: Subscriber<R>,
  69               private notifier: (notifications: Observable<any>) => Observable<any>,
  70               private source: Observable<T>) {
  71     super(destination);
  72   }
  73
  74   notifyNext(outerValue: T, innerValue: R,
  75              outerIndex: number, innerIndex: number,
  76              innerSub: InnerSubscriber<T, R>): void {
  77     this.sourceIsBeingSubscribedTo = true;
  78     this.source.subscribe(this);
  79   }
  80
  81   notifyComplete(innerSub: InnerSubscriber<T, R>): void {
  82     if (this.sourceIsBeingSubscribedTo === false) {
  83       return super.complete();
  84     }
  85   }
  86
  87   complete() {
  88     this.sourceIsBeingSubscribedTo = false;
  89
  90     if (!this.isStopped) {
  91       if (!this.retries) {
  92         this.subscribeToRetries();
  93       }
  94       if (!this.retriesSubscription || this.retriesSubscription.closed) {
  95         return super.complete();
  96       }
  97
  98       this._unsubscribeAndRecycle();
  99       this.notifications.next();
 100     }
 101   }
 102
 103   /** @deprecated This is an internal implementation detail, do not use. */
 104   _unsubscribe() {
 105     const { notifications, retriesSubscription } = this;
 106     if (notifications) {
 107       notifications.unsubscribe();
 108       this.notifications = null;
 109     }
 110     if (retriesSubscription) {
 111       retriesSubscription.unsubscribe();
 112       this.retriesSubscription = null;
 113     }
 114     this.retries = null;
 115   }
 116
 117   /** @deprecated This is an internal implementation detail, do not use. */
 118   _unsubscribeAndRecycle(): Subscriber<T> {
 119     const { _unsubscribe } = this;
 120
 121     this._unsubscribe = null;
 122     super._unsubscribeAndRecycle();
 123     this._unsubscribe = _unsubscribe;
 124
 125     return this;
 126   }
 127
 128   private subscribeToRetries() {
 129     this.notifications = new Subject();
 130     let retries;
 131     try {
 132       const { notifier } = this;
 133       retries = notifier(this.notifications);
 134     } catch (e) {
 135       return super.complete();
 136     }
 137     this.retries = retries;
 138     this.retriesSubscription = subscribeToResult(this, retries);
 139   }
 140 }