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