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

   1 import { Observable } from '../Observable';
   2 import { OperatorFunction } from '../types';
   3 import { Subject } from '../Subject';
   4 import { Subscriber } from '../Subscriber';
   5 import { Operator } from '../Operator';
   6 import { SimpleOuterSubscriber, innerSubscribe, SimpleInnerSubscriber } from '../innerSubscribe';
   7
   8 /**
   9  * Branch out the source Observable values as a nested Observable whenever
  10  * `windowBoundaries` emits.
  11  *
  12  * <span class="informal">It's like {@link buffer}, but emits a nested Observable
  13  * instead of an array.</span>
  14  *
  15  * ![](window.png)
  16  *
  17  * Returns an Observable that emits windows of items it collects from the source
  18  * Observable. The output Observable emits connected, non-overlapping
  19  * windows. It emits the current window and opens a new one whenever the
  20  * Observable `windowBoundaries` emits an item. Because each window is an
  21  * Observable, the output is a higher-order Observable.
  22  *
  23  * ## Example
  24  * In every window of 1 second each, emit at most 2 click events
  25  * ```ts
  26  * import { fromEvent, interval } from 'rxjs';
  27  * import { window, mergeAll, map, take } from 'rxjs/operators';
  28  *
  29  *  const clicks = fromEvent(document, 'click');
  30  *  const sec = interval(1000);
  31  *  const result = clicks.pipe(
  32  *      window(sec),
  33  *      map(win => win.pipe(take(2))), // each window has at most 2 emissions
  34  *      mergeAll(),              // flatten the Observable-of-Observables
  35  *  );
  36  *  result.subscribe(x => console.log(x));
  37  * ```
  38  * @see {@link windowCount}
  39  * @see {@link windowTime}
  40  * @see {@link windowToggle}
  41  * @see {@link windowWhen}
  42  * @see {@link buffer}
  43  *
  44  * @param {Observable<any>} windowBoundaries An Observable that completes the
  45  * previous window and starts a new window.
  46  * @return {Observable<Observable<T>>} An Observable of windows, which are
  47  * Observables emitting values of the source Observable.
  48  * @method window
  49  * @owner Observable
  50  */
  51 export function window<T>(windowBoundaries: Observable<any>): OperatorFunction<T, Observable<T>> {
  52   return function windowOperatorFunction(source: Observable<T>) {
  53     return source.lift(new WindowOperator(windowBoundaries));
  54   };
  55 }
  56
  57 class WindowOperator<T> implements Operator<T, Observable<T>> {
  58
  59   constructor(private windowBoundaries: Observable<any>) {
  60   }
  61
  62   call(subscriber: Subscriber<Observable<T>>, source: any): any {
  63     const windowSubscriber = new WindowSubscriber(subscriber);
  64     const sourceSubscription = source.subscribe(windowSubscriber);
  65     if (!sourceSubscription.closed) {
  66       windowSubscriber.add(innerSubscribe(this.windowBoundaries, new SimpleInnerSubscriber(windowSubscriber)));
  67     }
  68     return sourceSubscription;
  69   }
  70 }
  71
  72 /**
  73  * We need this JSDoc comment for affecting ESDoc.
  74  * @ignore
  75  * @extends {Ignored}
  76  */
  77 class WindowSubscriber<T> extends SimpleOuterSubscriber<T, any> {
  78
  79   private window: Subject<T> = new Subject<T>();
  80
  81   constructor(destination: Subscriber<Observable<T>>) {
  82     super(destination);
  83     destination.next(this.window);
  84   }
  85
  86   notifyNext(): void {
  87     this.openWindow();
  88   }
  89
  90   notifyError(error: any): void {
  91     this._error(error);
  92   }
  93
  94   notifyComplete(): void {
  95     this._complete();
  96   }
  97
  98   protected _next(value: T): void {
  99     this.window.next(value);
 100   }
 101
 102   protected _error(err: any): void {
 103     this.window.error(err);
 104     this.destination.error!(err);
 105   }
 106
 107   protected _complete(): void {
 108     this.window.complete();
 109     this.destination.complete!();
 110   }
 111
 112   /** @deprecated This is an internal implementation detail, do not use. */
 113   _unsubscribe() {
 114     this.window = null!;
 115   }
 116
 117   private openWindow(): void  {
 118     const prevWindow = this.window;
 119     if (prevWindow) {
 120       prevWindow.complete();
 121     }
 122     const destination = this.destination;
 123     const newWindow = this.window = new Subject<T>();
 124     destination.next!(newWindow);
 125   }
 126 }