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

   1 import { Observable } from '../Observable';
   2 import { ReplaySubject } from '../ReplaySubject';
   3 import { Subscription } from '../Subscription';
   4 import { MonoTypeOperatorFunction, SchedulerLike } from '../types';
   5 import { Subscriber } from '../Subscriber';
   6
   7 export interface ShareReplayConfig {
   8   bufferSize?: number;
   9   windowTime?: number;
  10   refCount: boolean;
  11   scheduler?: SchedulerLike;
  12 }
  13
  14 /**
  15  * Share source and replay specified number of emissions on subscription.
  16  *
  17  * This operator is a specialization of `replay` that connects to a source observable
  18  * and multicasts through a `ReplaySubject` constructed with the specified arguments.
  19  * A successfully completed source will stay cached in the `shareReplayed observable` forever,
  20  * but an errored source can be retried.
  21  *
  22  * ## Why use shareReplay?
  23  * You generally want to use `shareReplay` when you have side-effects or taxing computations
  24  * that you do not wish to be executed amongst multiple subscribers.
  25  * It may also be valuable in situations where you know you will have late subscribers to
  26  * a stream that need access to previously emitted values.
  27  * This ability to replay values on subscription is what differentiates {@link share} and `shareReplay`.
  28  *
  29  * ![](shareReplay.png)
  30  *
  31  * ## Example
  32  * ```ts
  33  * import { interval } from 'rxjs';
  34  * import { shareReplay, take } from 'rxjs/operators';
  35  *
  36  * const obs$ = interval(1000);
  37  * const shared$ = obs$.pipe(
  38  *   take(4),
  39  *   shareReplay(3)
  40  * );
  41  * shared$.subscribe(x => console.log('source A: ', x));
  42  * shared$.subscribe(y => console.log('source B: ', y));
  43  *
  44  * ```
  45  *
  46  * @see {@link publish}
  47  * @see {@link share}
  48  * @see {@link publishReplay}
  49  *
  50  * @param {Number} [bufferSize=Number.POSITIVE_INFINITY] Maximum element count of the replay buffer.
  51  * @param {Number} [windowTime=Number.POSITIVE_INFINITY] Maximum time length of the replay buffer in milliseconds.
  52  * @param {Scheduler} [scheduler] Scheduler where connected observers within the selector function
  53  * will be invoked on.
  54  * @return {Observable} An observable sequence that contains the elements of a sequence produced
  55  * by multicasting the source sequence within a selector function.
  56  * @method shareReplay
  57  * @owner Observable
  58  */
  59 export function shareReplay<T>(config: ShareReplayConfig): MonoTypeOperatorFunction<T>;
  60 export function shareReplay<T>(bufferSize?: number, windowTime?: number, scheduler?: SchedulerLike): MonoTypeOperatorFunction<T>;
  61 export function shareReplay<T>(
  62   configOrBufferSize?: ShareReplayConfig | number,
  63   windowTime?: number,
  64   scheduler?: SchedulerLike
  65 ): MonoTypeOperatorFunction<T> {
  66   let config: ShareReplayConfig;
  67   if (configOrBufferSize && typeof configOrBufferSize === 'object') {
  68     config = configOrBufferSize as ShareReplayConfig;
  69   } else {
  70     config = {
  71       bufferSize: configOrBufferSize as number | undefined,
  72       windowTime,
  73       refCount: false,
  74       scheduler
  75     };
  76   }
  77   return (source: Observable<T>) => source.lift(shareReplayOperator(config));
  78 }
  79
  80 function shareReplayOperator<T>({
  81   bufferSize = Number.POSITIVE_INFINITY,
  82   windowTime = Number.POSITIVE_INFINITY,
  83   refCount: useRefCount,
  84   scheduler
  85 }: ShareReplayConfig) {
  86   let subject: ReplaySubject<T> | undefined;
  87   let refCount = 0;
  88   let subscription: Subscription | undefined;
  89   let hasError = false;
  90   let isComplete = false;
  91
  92   return function shareReplayOperation(this: Subscriber<T>, source: Observable<T>) {
  93     refCount++;
  94     let innerSub: Subscription;
  95     if (!subject || hasError) {
  96       hasError = false;
  97       subject = new ReplaySubject<T>(bufferSize, windowTime, scheduler);
  98       innerSub = subject.subscribe(this);
  99       subscription = source.subscribe({
 100         next(value) { subject.next(value); },
 101         error(err) {
 102           hasError = true;
 103           subject.error(err);
 104         },
 105         complete() {
 106           isComplete = true;
 107           subscription = undefined;
 108           subject.complete();
 109         },
 110       });
 111     } else {
 112       innerSub = subject.subscribe(this);
 113     }
 114
 115     this.add(() => {
 116       refCount--;
 117       innerSub.unsubscribe();
 118       if (subscription && !isComplete && useRefCount && refCount === 0) {
 119         subscription.unsubscribe();
 120         subscription = undefined;
 121         subject = undefined;
 122       }
 123     });
 124   };
 125 }