bufferTime.js
2.59 KB
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
import { Subscription } from '../Subscription';
import { operate } from '../util/lift';
import { createOperatorSubscriber } from './OperatorSubscriber';
import { arrRemove } from '../util/arrRemove';
import { asyncScheduler } from '../scheduler/async';
import { popScheduler } from '../util/args';
import { executeSchedule } from '../util/executeSchedule';
export function bufferTime(bufferTimeSpan, ...otherArgs) {
var _a, _b;
const scheduler = (_a = popScheduler(otherArgs)) !== null && _a !== void 0 ? _a : asyncScheduler;
const bufferCreationInterval = (_b = otherArgs[0]) !== null && _b !== void 0 ? _b : null;
const maxBufferSize = otherArgs[1] || Infinity;
return operate((source, subscriber) => {
let bufferRecords = [];
let restartOnEmit = false;
const emit = (record) => {
const { buffer, subs } = record;
subs.unsubscribe();
arrRemove(bufferRecords, record);
subscriber.next(buffer);
restartOnEmit && startBuffer();
};
const startBuffer = () => {
if (bufferRecords) {
const subs = new Subscription();
subscriber.add(subs);
const buffer = [];
const record = {
buffer,
subs,
};
bufferRecords.push(record);
executeSchedule(subs, scheduler, () => emit(record), bufferTimeSpan);
}
};
if (bufferCreationInterval !== null && bufferCreationInterval >= 0) {
executeSchedule(subscriber, scheduler, startBuffer, bufferCreationInterval, true);
}
else {
restartOnEmit = true;
}
startBuffer();
const bufferTimeSubscriber = createOperatorSubscriber(subscriber, (value) => {
const recordsCopy = bufferRecords.slice();
for (const record of recordsCopy) {
const { buffer } = record;
buffer.push(value);
maxBufferSize <= buffer.length && emit(record);
}
}, () => {
while (bufferRecords === null || bufferRecords === void 0 ? void 0 : bufferRecords.length) {
subscriber.next(bufferRecords.shift().buffer);
}
bufferTimeSubscriber === null || bufferTimeSubscriber === void 0 ? void 0 : bufferTimeSubscriber.unsubscribe();
subscriber.complete();
subscriber.unsubscribe();
}, undefined, () => (bufferRecords = null));
source.subscribe(bufferTimeSubscriber);
});
}
//# sourceMappingURL=bufferTime.js.map