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
"use strict";
var __extends = (this && this.__extends) || function (d, b) {
for (var p in b) if (b.hasOwnProperty(p)) d[p] = b[p];
function __() { this.constructor = d; }
d.prototype = b === null ? Object.create(b) : (__.prototype = b.prototype, new __());
};
var Subscriber_1 = require('../Subscriber');
/**
* 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>
*
* <img src="./img/bufferCount.png" width="100%">
*
* 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.
*
* @example <caption>Emit the last two click events as an array</caption>
* var clicks = Rx.Observable.fromEvent(document, 'click');
* var buffered = clicks.bufferCount(2);
* buffered.subscribe(x => console.log(x));
*
* @example <caption>On every click, emit the last two click events as an array</caption>
* var clicks = Rx.Observable.fromEvent(document, 'click');
* var buffered = clicks.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
*/
function bufferCount(bufferSize, startBufferEvery) {
if (startBufferEvery === void 0) { startBufferEvery = null; }
return function bufferCountOperatorFunction(source) {
return source.lift(new BufferCountOperator(bufferSize, startBufferEvery));
};
}
exports.bufferCount = bufferCount;
var BufferCountOperator = (function () {
function BufferCountOperator(bufferSize, startBufferEvery) {
this.bufferSize = bufferSize;
this.startBufferEvery = startBufferEvery;
if (!startBufferEvery || bufferSize === startBufferEvery) {
this.subscriberClass = BufferCountSubscriber;
}
else {
this.subscriberClass = BufferSkipCountSubscriber;
}
}
BufferCountOperator.prototype.call = function (subscriber, source) {
return source.subscribe(new this.subscriberClass(subscriber, this.bufferSize, this.startBufferEvery));
};
return BufferCountOperator;
}());
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
var BufferCountSubscriber = (function (_super) {
__extends(BufferCountSubscriber, _super);
function BufferCountSubscriber(destination, bufferSize) {
_super.call(this, destination);
this.bufferSize = bufferSize;
this.buffer = [];
}
BufferCountSubscriber.prototype._next = function (value) {
var buffer = this.buffer;
buffer.push(value);
if (buffer.length == this.bufferSize) {
this.destination.next(buffer);
this.buffer = [];
}
};
BufferCountSubscriber.prototype._complete = function () {
var buffer = this.buffer;
if (buffer.length > 0) {
this.destination.next(buffer);
}
_super.prototype._complete.call(this);
};
return BufferCountSubscriber;
}(Subscriber_1.Subscriber));
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
var BufferSkipCountSubscriber = (function (_super) {
__extends(BufferSkipCountSubscriber, _super);
function BufferSkipCountSubscriber(destination, bufferSize, startBufferEvery) {
_super.call(this, destination);
this.bufferSize = bufferSize;
this.startBufferEvery = startBufferEvery;
this.buffers = [];
this.count = 0;
}
BufferSkipCountSubscriber.prototype._next = function (value) {
var _a = this, bufferSize = _a.bufferSize, startBufferEvery = _a.startBufferEvery, buffers = _a.buffers, count = _a.count;
this.count++;
if (count % startBufferEvery === 0) {
buffers.push([]);
}
for (var i = buffers.length; i--;) {
var buffer = buffers[i];
buffer.push(value);
if (buffer.length === bufferSize) {
buffers.splice(i, 1);
this.destination.next(buffer);
}
}
};
BufferSkipCountSubscriber.prototype._complete = function () {
var _a = this, buffers = _a.buffers, destination = _a.destination;
while (buffers.length > 0) {
var buffer = buffers.shift();
if (buffer.length > 0) {
destination.next(buffer);
}
}
_super.prototype._complete.call(this);
};
return BufferSkipCountSubscriber;
}(Subscriber_1.Subscriber));
//# sourceMappingURL=bufferCount.js.map