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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
import { Operator } from '../Operator';
import { Subscriber } from '../Subscriber';
import { Observable } from '../Observable';
import { OperatorFunction, TeardownLogic } from '../types';
/**
* Buffers the source Observable values until the size hits the maximum
* `bufferSize` given.
*
* <span class="informal">Collects values from the past as an array, and emits
* that array only when its size reaches `bufferSize`.</span>
*
* 
*
* Buffers a number of values from the source Observable by `bufferSize` then
* emits the buffer and clears it, and starts a new buffer each
* `startBufferEvery` values. If `startBufferEvery` is not provided or is
* `null`, then new buffers are started immediately at the start of the source
* and when each buffer closes and is emitted.
*
* ## Examples
*
* Emit the last two click events as an array
*
* ```javascript
* import { fromEvent } from 'rxjs';
* import { bufferCount } from 'rxjs/operators';
*
* const clicks = fromEvent(document, 'click');
* const buffered = clicks.pipe(bufferCount(2));
* buffered.subscribe(x => console.log(x));
* ```
*
* On every click, emit the last two click events as an array
*
* ```javascript
* import { fromEvent } from 'rxjs';
* import { bufferCount } from 'rxjs/operators';
*
* const clicks = fromEvent(document, 'click');
* const buffered = clicks.pipe(bufferCount(2, 1));
* buffered.subscribe(x => console.log(x));
* ```
*
* @see {@link buffer}
* @see {@link bufferTime}
* @see {@link bufferToggle}
* @see {@link bufferWhen}
* @see {@link pairwise}
* @see {@link windowCount}
*
* @param {number} bufferSize The maximum size of the buffer emitted.
* @param {number} [startBufferEvery] Interval at which to start a new buffer.
* For example if `startBufferEvery` is `2`, then a new buffer will be started
* on every other value from the source. A new buffer is started at the
* beginning of the source by default.
* @return {Observable<T[]>} An Observable of arrays of buffered values.
* @method bufferCount
* @owner Observable
*/
export function bufferCount<T>(bufferSize: number, startBufferEvery: number = null): OperatorFunction<T, T[]> {
return function bufferCountOperatorFunction(source: Observable<T>) {
return source.lift(new BufferCountOperator<T>(bufferSize, startBufferEvery));
};
}
class BufferCountOperator<T> implements Operator<T, T[]> {
private subscriberClass: any;
constructor(private bufferSize: number, private startBufferEvery: number) {
if (!startBufferEvery || bufferSize === startBufferEvery) {
this.subscriberClass = BufferCountSubscriber;
} else {
this.subscriberClass = BufferSkipCountSubscriber;
}
}
call(subscriber: Subscriber<T[]>, source: any): TeardownLogic {
return source.subscribe(new this.subscriberClass(subscriber, this.bufferSize, this.startBufferEvery));
}
}
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class BufferCountSubscriber<T> extends Subscriber<T> {
private buffer: T[] = [];
constructor(destination: Subscriber<T[]>, private bufferSize: number) {
super(destination);
}
protected _next(value: T): void {
const buffer = this.buffer;
buffer.push(value);
if (buffer.length == this.bufferSize) {
this.destination.next(buffer);
this.buffer = [];
}
}
protected _complete(): void {
const buffer = this.buffer;
if (buffer.length > 0) {
this.destination.next(buffer);
}
super._complete();
}
}
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
class BufferSkipCountSubscriber<T> extends Subscriber<T> {
private buffers: Array<T[]> = [];
private count: number = 0;
constructor(destination: Subscriber<T[]>, private bufferSize: number, private startBufferEvery: number) {
super(destination);
}
protected _next(value: T): void {
const { bufferSize, startBufferEvery, buffers, count } = this;
this.count++;
if (count % startBufferEvery === 0) {
buffers.push([]);
}
for (let i = buffers.length; i--; ) {
const buffer = buffers[i];
buffer.push(value);
if (buffer.length === bufferSize) {
buffers.splice(i, 1);
this.destination.next(buffer);
}
}
}
protected _complete(): void {
const { buffers, destination } = this;
while (buffers.length > 0) {
let buffer = buffers.shift();
if (buffer.length > 0) {
destination.next(buffer);
}
}
super._complete();
}
}
|
