bufferToggle.js
6.11 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
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
"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 Subscription_1 = require('../Subscription');
var subscribeToResult_1 = require('../util/subscribeToResult');
var OuterSubscriber_1 = require('../OuterSubscriber');
/**
* Buffers the source Observable values starting from an emission from
* `openings` and ending when the output of `closingSelector` emits.
*
* <span class="informal">Collects values from the past as an array. Starts
* collecting only when `opening` emits, and calls the `closingSelector`
* function to get an Observable that tells when to close the buffer.</span>
*
* <img src="./img/bufferToggle.png" width="100%">
*
* Buffers values from the source by opening the buffer via signals from an
* Observable provided to `openings`, and closing and sending the buffers when
* a Subscribable or Promise returned by the `closingSelector` function emits.
*
* @example <caption>Every other second, emit the click events from the next 500ms</caption>
* var clicks = Rx.Observable.fromEvent(document, 'click');
* var openings = Rx.Observable.interval(1000);
* var buffered = clicks.bufferToggle(openings, i =>
* i % 2 ? Rx.Observable.interval(500) : Rx.Observable.empty()
* );
* buffered.subscribe(x => console.log(x));
*
* @see {@link buffer}
* @see {@link bufferCount}
* @see {@link bufferTime}
* @see {@link bufferWhen}
* @see {@link windowToggle}
*
* @param {SubscribableOrPromise<O>} openings A Subscribable or Promise of notifications to start new
* buffers.
* @param {function(value: O): SubscribableOrPromise} closingSelector A function that takes
* the value emitted by the `openings` observable and returns a Subscribable or Promise,
* which, when it emits, signals that the associated buffer should be emitted
* and cleared.
* @return {Observable<T[]>} An observable of arrays of buffered values.
* @method bufferToggle
* @owner Observable
*/
function bufferToggle(openings, closingSelector) {
return this.lift(new BufferToggleOperator(openings, closingSelector));
}
exports.bufferToggle = bufferToggle;
var BufferToggleOperator = (function () {
function BufferToggleOperator(openings, closingSelector) {
this.openings = openings;
this.closingSelector = closingSelector;
}
BufferToggleOperator.prototype.call = function (subscriber, source) {
return source.subscribe(new BufferToggleSubscriber(subscriber, this.openings, this.closingSelector));
};
return BufferToggleOperator;
}());
/**
* We need this JSDoc comment for affecting ESDoc.
* @ignore
* @extends {Ignored}
*/
var BufferToggleSubscriber = (function (_super) {
__extends(BufferToggleSubscriber, _super);
function BufferToggleSubscriber(destination, openings, closingSelector) {
_super.call(this, destination);
this.openings = openings;
this.closingSelector = closingSelector;
this.contexts = [];
this.add(subscribeToResult_1.subscribeToResult(this, openings));
}
BufferToggleSubscriber.prototype._next = function (value) {
var contexts = this.contexts;
var len = contexts.length;
for (var i = 0; i < len; i++) {
contexts[i].buffer.push(value);
}
};
BufferToggleSubscriber.prototype._error = function (err) {
var contexts = this.contexts;
while (contexts.length > 0) {
var context = contexts.shift();
context.subscription.unsubscribe();
context.buffer = null;
context.subscription = null;
}
this.contexts = null;
_super.prototype._error.call(this, err);
};
BufferToggleSubscriber.prototype._complete = function () {
var contexts = this.contexts;
while (contexts.length > 0) {
var context = contexts.shift();
this.destination.next(context.buffer);
context.subscription.unsubscribe();
context.buffer = null;
context.subscription = null;
}
this.contexts = null;
_super.prototype._complete.call(this);
};
BufferToggleSubscriber.prototype.notifyNext = function (outerValue, innerValue, outerIndex, innerIndex, innerSub) {
outerValue ? this.closeBuffer(outerValue) : this.openBuffer(innerValue);
};
BufferToggleSubscriber.prototype.notifyComplete = function (innerSub) {
this.closeBuffer(innerSub.context);
};
BufferToggleSubscriber.prototype.openBuffer = function (value) {
try {
var closingSelector = this.closingSelector;
var closingNotifier = closingSelector.call(this, value);
if (closingNotifier) {
this.trySubscribe(closingNotifier);
}
}
catch (err) {
this._error(err);
}
};
BufferToggleSubscriber.prototype.closeBuffer = function (context) {
var contexts = this.contexts;
if (contexts && context) {
var buffer = context.buffer, subscription = context.subscription;
this.destination.next(buffer);
contexts.splice(contexts.indexOf(context), 1);
this.remove(subscription);
subscription.unsubscribe();
}
};
BufferToggleSubscriber.prototype.trySubscribe = function (closingNotifier) {
var contexts = this.contexts;
var buffer = [];
var subscription = new Subscription_1.Subscription();
var context = { buffer: buffer, subscription: subscription };
contexts.push(context);
var innerSubscription = subscribeToResult_1.subscribeToResult(this, closingNotifier, context);
if (!innerSubscription || innerSubscription.closed) {
this.closeBuffer(context);
}
else {
innerSubscription.context = context;
this.add(innerSubscription);
subscription.add(innerSubscription);
}
};
return BufferToggleSubscriber;
}(OuterSubscriber_1.OuterSubscriber));
//# sourceMappingURL=bufferToggle.js.map