144 lines
5.2 KiB
JavaScript
144 lines
5.2 KiB
JavaScript
import { Subscription } from '../Subscription';
|
|
import { subscribeToResult } from '../util/subscribeToResult';
|
|
import { OuterSubscriber } from '../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
|
|
*/
|
|
export function bufferToggle(openings, closingSelector) {
|
|
return function bufferToggleOperatorFunction(source) {
|
|
return source.lift(new BufferToggleOperator(openings, closingSelector));
|
|
};
|
|
}
|
|
class BufferToggleOperator {
|
|
constructor(openings, closingSelector) {
|
|
this.openings = openings;
|
|
this.closingSelector = closingSelector;
|
|
}
|
|
call(subscriber, source) {
|
|
return source.subscribe(new BufferToggleSubscriber(subscriber, this.openings, this.closingSelector));
|
|
}
|
|
}
|
|
/**
|
|
* We need this JSDoc comment for affecting ESDoc.
|
|
* @ignore
|
|
* @extends {Ignored}
|
|
*/
|
|
class BufferToggleSubscriber extends OuterSubscriber {
|
|
constructor(destination, openings, closingSelector) {
|
|
super(destination);
|
|
this.openings = openings;
|
|
this.closingSelector = closingSelector;
|
|
this.contexts = [];
|
|
this.add(subscribeToResult(this, openings));
|
|
}
|
|
_next(value) {
|
|
const contexts = this.contexts;
|
|
const len = contexts.length;
|
|
for (let i = 0; i < len; i++) {
|
|
contexts[i].buffer.push(value);
|
|
}
|
|
}
|
|
_error(err) {
|
|
const contexts = this.contexts;
|
|
while (contexts.length > 0) {
|
|
const context = contexts.shift();
|
|
context.subscription.unsubscribe();
|
|
context.buffer = null;
|
|
context.subscription = null;
|
|
}
|
|
this.contexts = null;
|
|
super._error(err);
|
|
}
|
|
_complete() {
|
|
const contexts = this.contexts;
|
|
while (contexts.length > 0) {
|
|
const context = contexts.shift();
|
|
this.destination.next(context.buffer);
|
|
context.subscription.unsubscribe();
|
|
context.buffer = null;
|
|
context.subscription = null;
|
|
}
|
|
this.contexts = null;
|
|
super._complete();
|
|
}
|
|
notifyNext(outerValue, innerValue, outerIndex, innerIndex, innerSub) {
|
|
outerValue ? this.closeBuffer(outerValue) : this.openBuffer(innerValue);
|
|
}
|
|
notifyComplete(innerSub) {
|
|
this.closeBuffer(innerSub.context);
|
|
}
|
|
openBuffer(value) {
|
|
try {
|
|
const closingSelector = this.closingSelector;
|
|
const closingNotifier = closingSelector.call(this, value);
|
|
if (closingNotifier) {
|
|
this.trySubscribe(closingNotifier);
|
|
}
|
|
}
|
|
catch (err) {
|
|
this._error(err);
|
|
}
|
|
}
|
|
closeBuffer(context) {
|
|
const contexts = this.contexts;
|
|
if (contexts && context) {
|
|
const { buffer, subscription } = context;
|
|
this.destination.next(buffer);
|
|
contexts.splice(contexts.indexOf(context), 1);
|
|
this.remove(subscription);
|
|
subscription.unsubscribe();
|
|
}
|
|
}
|
|
trySubscribe(closingNotifier) {
|
|
const contexts = this.contexts;
|
|
const buffer = [];
|
|
const subscription = new Subscription();
|
|
const context = { buffer, subscription };
|
|
contexts.push(context);
|
|
const innerSubscription = subscribeToResult(this, closingNotifier, context);
|
|
if (!innerSubscription || innerSubscription.closed) {
|
|
this.closeBuffer(context);
|
|
}
|
|
else {
|
|
innerSubscription.context = context;
|
|
this.add(innerSubscription);
|
|
subscription.add(innerSubscription);
|
|
}
|
|
}
|
|
}
|
|
//# sourceMappingURL=bufferToggle.js.map
|