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

   1 import { Operator } from '../Operator';
   2 import { Subscriber } from '../Subscriber';
   3 import { Observable } from '../Observable';
   4 import { OuterSubscriber } from '../OuterSubscriber';
   5 import { InnerSubscriber } from '../InnerSubscriber';
   6 import { subscribeToResult } from '../util/subscribeToResult';
   7 import { OperatorFunction } from '../types';
   8
   9 /**
  10  * Buffers the source Observable values until `closingNotifier` emits.
  11  *
  12  * <span class="informal">Collects values from the past as an array, and emits
  13  * that array only when another Observable emits.</span>
  14  *
  15  * ![](buffer.png)
  16  *
  17  * Buffers the incoming Observable values until the given `closingNotifier`
  18  * Observable emits a value, at which point it emits the buffer on the output
  19  * Observable and starts a new buffer internally, awaiting the next time
  20  * `closingNotifier` emits.
  21  *
  22  * ## Example
  23  *
  24  * On every click, emit array of most recent interval events
  25  *
  26  * ```ts
  27  * import { fromEvent, interval } from 'rxjs';
  28  * import { buffer } from 'rxjs/operators';
  29  *
  30  * const clicks = fromEvent(document, 'click');
  31  * const intervalEvents = interval(1000);
  32  * const buffered = intervalEvents.pipe(buffer(clicks));
  33  * buffered.subscribe(x => console.log(x));
  34  * ```
  35  *
  36  * @see {@link bufferCount}
  37  * @see {@link bufferTime}
  38  * @see {@link bufferToggle}
  39  * @see {@link bufferWhen}
  40  * @see {@link window}
  41  *
  42  * @param {Observable<any>} closingNotifier An Observable that signals the
  43  * buffer to be emitted on the output Observable.
  44  * @return {Observable<T[]>} An Observable of buffers, which are arrays of
  45  * values.
  46  * @method buffer
  47  * @owner Observable
  48  */
  49 export function buffer<T>(closingNotifier: Observable<any>): OperatorFunction<T, T[]> {
  50   return function bufferOperatorFunction(source: Observable<T>) {
  51     return source.lift(new BufferOperator<T>(closingNotifier));
  52   };
  53 }
  54
  55 class BufferOperator<T> implements Operator<T, T[]> {
  56
  57   constructor(private closingNotifier: Observable<any>) {
  58   }
  59
  60   call(subscriber: Subscriber<T[]>, source: any): any {
  61     return source.subscribe(new BufferSubscriber(subscriber, this.closingNotifier));
  62   }
  63 }
  64
  65 /**
  66  * We need this JSDoc comment for affecting ESDoc.
  67  * @ignore
  68  * @extends {Ignored}
  69  */
  70 class BufferSubscriber<T> extends OuterSubscriber<T, any> {
  71   private buffer: T[] = [];
  72
  73   constructor(destination: Subscriber<T[]>, closingNotifier: Observable<any>) {
  74     super(destination);
  75     this.add(subscribeToResult(this, closingNotifier));
  76   }
  77
  78   protected _next(value: T) {
  79     this.buffer.push(value);
  80   }
  81
  82   notifyNext(outerValue: T, innerValue: any,
  83              outerIndex: number, innerIndex: number,
  84              innerSub: InnerSubscriber<T, any>): void {
  85     const buffer = this.buffer;
  86     this.buffer = [];
  87     this.destination.next(buffer);
  88   }
  89 }