-
Notifications
You must be signed in to change notification settings - Fork 3k
/
buffer.ts
81 lines (77 loc) · 2.39 KB
/
buffer.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
import { OperatorFunction, ObservableInput } from '../types';
import { operate } from '../util/lift';
import { noop } from '../util/noop';
import { createOperatorSubscriber } from './OperatorSubscriber';
import { from } from '../observable/from';
/**
* Buffers the source Observable values until `closingNotifier` emits.
*
* <span class="informal">Collects values from the past as an array, and emits
* that array only when another Observable emits.</span>
*
* ![](buffer.png)
*
* Buffers the incoming Observable values until the given `closingNotifier`
* `ObservableInput` (that internally gets converted to an Observable)
* emits a value, at which point it emits the buffer on the output
* Observable and starts a new buffer internally, awaiting the next time
* `closingNotifier` emits.
*
* ## Example
*
* On every click, emit array of most recent interval events
*
* ```ts
* import { fromEvent, interval, buffer } from 'rxjs';
*
* const clicks = fromEvent(document, 'click');
* const intervalEvents = interval(1000);
* const buffered = intervalEvents.pipe(buffer(clicks));
* buffered.subscribe(x => console.log(x));
* ```
*
* @see {@link bufferCount}
* @see {@link bufferTime}
* @see {@link bufferToggle}
* @see {@link bufferWhen}
* @see {@link window}
*
* @param closingNotifier An `ObservableInput` that signals the
* buffer to be emitted on the output Observable.
* @return A function that returns an Observable of buffers, which are arrays
* of values.
*/
export function buffer<T>(closingNotifier: ObservableInput<any>): OperatorFunction<T, T[]> {
return operate((source, subscriber) => {
// The current buffered values.
let currentBuffer: T[] = [];
// Subscribe to the closing notifier first.
from(closingNotifier).subscribe(
createOperatorSubscriber(
subscriber,
() => {
// Start a new buffer and emit the previous one.
const b = currentBuffer;
currentBuffer = [];
subscriber.next(b);
},
noop
)
);
// Subscribe to our source.
source.subscribe(
createOperatorSubscriber(
subscriber,
(value) => currentBuffer.push(value),
() => {
subscriber.next(currentBuffer);
subscriber.complete();
}
)
);
return () => {
// Ensure buffered values are released on finalization.
currentBuffer = null!;
};
});
}