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';
8 * Buffers the source Observable values until `closingNotifier` emits.
10 * <span class="informal">Collects values from the past as an array, and emits
11 * that array only when another Observable emits.</span>
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.
22 * On every click, emit array of most recent interval events
25 * import { fromEvent, interval } from 'rxjs';
26 * import { buffer } from 'rxjs/operators';
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));
34 * @see {@link bufferCount}
35 * @see {@link bufferTime}
36 * @see {@link bufferToggle}
37 * @see {@link bufferWhen}
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
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));
53 class BufferOperator<T> implements Operator<T, T[]> {
55 constructor(private closingNotifier: Observable<any>) {
58 call(subscriber: Subscriber<T[]>, source: any): any {
59 return source.subscribe(new BufferSubscriber(subscriber, this.closingNotifier));
64 * We need this JSDoc comment for affecting ESDoc.
68 class BufferSubscriber<T> extends SimpleOuterSubscriber<T, any> {
69 private buffer: T[] = [];
71 constructor(destination: Subscriber<T[]>, closingNotifier: Observable<any>) {
73 this.add(innerSubscribe(closingNotifier, new SimpleInnerSubscriber(this)));
76 protected _next(value: T) {
77 this.buffer.push(value);
81 const buffer = this.buffer;
83 this.destination.next!(buffer);