"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 Subject_1 = require('../Subject'); var Subscription_1 = require('../Subscription'); var tryCatch_1 = require('../util/tryCatch'); var errorObject_1 = require('../util/errorObject'); var OuterSubscriber_1 = require('../OuterSubscriber'); var subscribeToResult_1 = require('../util/subscribeToResult'); /** * Branch out the source Observable values as a nested Observable starting from * an emission from `openings` and ending when the output of `closingSelector` * emits. * * It's like {@link bufferToggle}, but emits a nested * Observable instead of an array. * * * * Returns an Observable that emits windows of items it collects from the source * Observable. The output Observable emits windows that contain those items * emitted by the source Observable between the time when the `openings` * Observable emits an item and when the Observable returned by * `closingSelector` emits an item. * * @example Every other second, emit the click events from the next 500ms * var clicks = Rx.Observable.fromEvent(document, 'click'); * var openings = Rx.Observable.interval(1000); * var result = clicks.windowToggle(openings, i => * i % 2 ? Rx.Observable.interval(500) : Rx.Observable.empty() * ).mergeAll(); * result.subscribe(x => console.log(x)); * * @see {@link window} * @see {@link windowCount} * @see {@link windowTime} * @see {@link windowWhen} * @see {@link bufferToggle} * * @param {Observable} openings An observable of notifications to start new * windows. * @param {function(value: O): Observable} closingSelector A function that takes * the value emitted by the `openings` observable and returns an Observable, * which, when it emits (either `next` or `complete`), signals that the * associated window should complete. * @return {Observable>} An observable of windows, which in turn * are Observables. * @method windowToggle * @owner Observable */ function windowToggle(openings, closingSelector) { return function (source) { return source.lift(new WindowToggleOperator(openings, closingSelector)); }; } exports.windowToggle = windowToggle; var WindowToggleOperator = (function () { function WindowToggleOperator(openings, closingSelector) { this.openings = openings; this.closingSelector = closingSelector; } WindowToggleOperator.prototype.call = function (subscriber, source) { return source.subscribe(new WindowToggleSubscriber(subscriber, this.openings, this.closingSelector)); }; return WindowToggleOperator; }()); /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ var WindowToggleSubscriber = (function (_super) { __extends(WindowToggleSubscriber, _super); function WindowToggleSubscriber(destination, openings, closingSelector) { _super.call(this, destination); this.openings = openings; this.closingSelector = closingSelector; this.contexts = []; this.add(this.openSubscription = subscribeToResult_1.subscribeToResult(this, openings, openings)); } WindowToggleSubscriber.prototype._next = function (value) { var contexts = this.contexts; if (contexts) { var len = contexts.length; for (var i = 0; i < len; i++) { contexts[i].window.next(value); } } }; WindowToggleSubscriber.prototype._error = function (err) { var contexts = this.contexts; this.contexts = null; if (contexts) { var len = contexts.length; var index = -1; while (++index < len) { var context = contexts[index]; context.window.error(err); context.subscription.unsubscribe(); } } _super.prototype._error.call(this, err); }; WindowToggleSubscriber.prototype._complete = function () { var contexts = this.contexts; this.contexts = null; if (contexts) { var len = contexts.length; var index = -1; while (++index < len) { var context = contexts[index]; context.window.complete(); context.subscription.unsubscribe(); } } _super.prototype._complete.call(this); }; /** @deprecated internal use only */ WindowToggleSubscriber.prototype._unsubscribe = function () { var contexts = this.contexts; this.contexts = null; if (contexts) { var len = contexts.length; var index = -1; while (++index < len) { var context = contexts[index]; context.window.unsubscribe(); context.subscription.unsubscribe(); } } }; WindowToggleSubscriber.prototype.notifyNext = function (outerValue, innerValue, outerIndex, innerIndex, innerSub) { if (outerValue === this.openings) { var closingSelector = this.closingSelector; var closingNotifier = tryCatch_1.tryCatch(closingSelector)(innerValue); if (closingNotifier === errorObject_1.errorObject) { return this.error(errorObject_1.errorObject.e); } else { var window_1 = new Subject_1.Subject(); var subscription = new Subscription_1.Subscription(); var context = { window: window_1, subscription: subscription }; this.contexts.push(context); var innerSubscription = subscribeToResult_1.subscribeToResult(this, closingNotifier, context); if (innerSubscription.closed) { this.closeWindow(this.contexts.length - 1); } else { innerSubscription.context = context; subscription.add(innerSubscription); } this.destination.next(window_1); } } else { this.closeWindow(this.contexts.indexOf(outerValue)); } }; WindowToggleSubscriber.prototype.notifyError = function (err) { this.error(err); }; WindowToggleSubscriber.prototype.notifyComplete = function (inner) { if (inner !== this.openSubscription) { this.closeWindow(this.contexts.indexOf(inner.context)); } }; WindowToggleSubscriber.prototype.closeWindow = function (index) { if (index === -1) { return; } var contexts = this.contexts; var context = contexts[index]; var window = context.window, subscription = context.subscription; contexts.splice(index, 1); window.complete(); subscription.unsubscribe(); }; return WindowToggleSubscriber; }(OuterSubscriber_1.OuterSubscriber)); //# sourceMappingURL=windowToggle.js.map