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

   1 import { Operator } from '../Operator';
   2 import { Subscriber } from '../Subscriber';
   3 import { Observable } from '../Observable';
   4 import { Subject } from '../Subject';
   5 import { OperatorFunction } from '../types';
   6
   7 /**
   8  * Branch out the source Observable values as a nested Observable with each
   9  * nested Observable emitting at most `windowSize` values.
  10  *
  11  * <span class="informal">It's like {@link bufferCount}, but emits a nested
  12  * Observable instead of an array.</span>
  13  *
  14  * ![](windowCount.png)
  15  *
  16  * Returns an Observable that emits windows of items it collects from the source
  17  * Observable. The output Observable emits windows every `startWindowEvery`
  18  * items, each containing no more than `windowSize` items. When the source
  19  * Observable completes or encounters an error, the output Observable emits
  20  * the current window and propagates the notification from the source
  21  * Observable. If `startWindowEvery` is not provided, then new windows are
  22  * started immediately at the start of the source and when each window completes
  23  * with size `windowSize`.
  24  *
  25  * ## Examples
  26  * Ignore every 3rd click event, starting from the first one
  27  * ```ts
  28  * import { fromEvent } from 'rxjs';
  29  * import { windowCount, map, mergeAll, skip } from 'rxjs/operators';
  30  *
  31  * const clicks = fromEvent(document, 'click');
  32  * const result = clicks.pipe(
  33  *   windowCount(3),
  34  *   map(win => win.pipe(skip(1))), // skip first of every 3 clicks
  35  *   mergeAll()                     // flatten the Observable-of-Observables
  36  * );
  37  * result.subscribe(x => console.log(x));
  38  * ```
  39  *
  40  * Ignore every 3rd click event, starting from the third one
  41  * ```ts
  42  * import { fromEvent } from 'rxjs';
  43  * import { windowCount, mergeAll } from 'rxjs/operators';
  44  *
  45  * const clicks = fromEvent(document, 'click');
  46  * const result = clicks.pipe(
  47  *   windowCount(2, 3),
  48  *   mergeAll(),              // flatten the Observable-of-Observables
  49  * );
  50  * result.subscribe(x => console.log(x));
  51  * ```
  52  *
  53  * @see {@link window}
  54  * @see {@link windowTime}
  55  * @see {@link windowToggle}
  56  * @see {@link windowWhen}
  57  * @see {@link bufferCount}
  58  *
  59  * @param {number} windowSize The maximum number of values emitted by each
  60  * window.
  61  * @param {number} [startWindowEvery] Interval at which to start a new window.
  62  * For example if `startWindowEvery` is `2`, then a new window will be started
  63  * on every other value from the source. A new window is started at the
  64  * beginning of the source by default.
  65  * @return {Observable<Observable<T>>} An Observable of windows, which in turn
  66  * are Observable of values.
  67  * @method windowCount
  68  * @owner Observable
  69  */
  70 export function windowCount<T>(windowSize: number,
  71                                startWindowEvery: number = 0): OperatorFunction<T, Observable<T>> {
  72   return function windowCountOperatorFunction(source: Observable<T>) {
  73     return source.lift(new WindowCountOperator<T>(windowSize, startWindowEvery));
  74   };
  75 }
  76
  77 class WindowCountOperator<T> implements Operator<T, Observable<T>> {
  78
  79   constructor(private windowSize: number,
  80               private startWindowEvery: number) {
  81   }
  82
  83   call(subscriber: Subscriber<Observable<T>>, source: any): any {
  84     return source.subscribe(new WindowCountSubscriber(subscriber, this.windowSize, this.startWindowEvery));
  85   }
  86 }
  87
  88 /**
  89  * We need this JSDoc comment for affecting ESDoc.
  90  * @ignore
  91  * @extends {Ignored}
  92  */
  93 class WindowCountSubscriber<T> extends Subscriber<T> {
  94   private windows: Subject<T>[] = [ new Subject<T>() ];
  95   private count: number = 0;
  96
  97   constructor(protected destination: Subscriber<Observable<T>>,
  98               private windowSize: number,
  99               private startWindowEvery: number) {
 100     super(destination);
 101     destination.next(this.windows[0]);
 102   }
 103
 104   protected _next(value: T) {
 105     const startWindowEvery = (this.startWindowEvery > 0) ? this.startWindowEvery : this.windowSize;
 106     const destination = this.destination;
 107     const windowSize = this.windowSize;
 108     const windows = this.windows;
 109     const len = windows.length;
 110
 111     for (let i = 0; i < len && !this.closed; i++) {
 112       windows[i].next(value);
 113     }
 114     const c = this.count - windowSize + 1;
 115     if (c >= 0 && c % startWindowEvery === 0 && !this.closed) {
 116       windows.shift().complete();
 117     }
 118     if (++this.count % startWindowEvery === 0 && !this.closed) {
 119       const window = new Subject<T>();
 120       windows.push(window);
 121       destination.next(window);
 122     }
 123   }
 124
 125   protected _error(err: any) {
 126     const windows = this.windows;
 127     if (windows) {
 128       while (windows.length > 0 && !this.closed) {
 129         windows.shift().error(err);
 130       }
 131     }
 132     this.destination.error(err);
 133   }
 134
 135   protected _complete() {
 136     const windows = this.windows;
 137     if (windows) {
 138       while (windows.length > 0 && !this.closed) {
 139         windows.shift().complete();
 140       }
 141     }
 142     this.destination.complete();
 143   }
 144
 145   protected _unsubscribe() {
 146     this.count = 0;
 147     this.windows = null;
 148   }
 149 }