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';
10 * Buffers the source Observable values until `closingNotifier` emits.
12 * <span class="informal">Collects values from the past as an array, and emits
13 * that array only when another Observable emits.</span>
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.
24 * On every click, emit array of most recent interval events
27 * import { fromEvent, interval } from 'rxjs';
28 * import { buffer } from 'rxjs/operators';
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));
36 * @see {@link bufferCount}
37 * @see {@link bufferTime}
38 * @see {@link bufferToggle}
39 * @see {@link bufferWhen}
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
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));
55 class BufferOperator<T> implements Operator<T, T[]> {
57 constructor(private closingNotifier: Observable<any>) {
60 call(subscriber: Subscriber<T[]>, source: any): any {
61 return source.subscribe(new BufferSubscriber(subscriber, this.closingNotifier));
66 * We need this JSDoc comment for affecting ESDoc.
70 class BufferSubscriber<T> extends OuterSubscriber<T, any> {
71 private buffer: T[] = [];
73 constructor(destination: Subscriber<T[]>, closingNotifier: Observable<any>) {
75 this.add(subscribeToResult(this, closingNotifier));
78 protected _next(value: T) {
79 this.buffer.push(value);
82 notifyNext(outerValue: T, innerValue: any,
83 outerIndex: number, innerIndex: number,
84 innerSub: InnerSubscriber<T, any>): void {
85 const buffer = this.buffer;
87 this.destination.next(buffer);