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