import { Observable } from '../Observable'; import { Operator } from '../Operator'; import { Subscriber } from '../Subscriber'; import { Subscription } from '../Subscription'; import { OuterSubscriber } from '../OuterSubscriber'; import { InnerSubscriber } from '../InnerSubscriber'; import { subscribeToResult } from '../util/subscribeToResult'; /** * Converts a higher-order Observable into a first-order Observable by * subscribing to only the most recently emitted of those inner Observables. * * Flattens an Observable-of-Observables by dropping the * previous inner Observable once a new one appears. * * * * `switch` subscribes to an Observable that emits Observables, also known as a * higher-order Observable. Each time it observes one of these emitted inner * Observables, the output Observable subscribes to the inner Observable and * begins emitting the items emitted by that. So far, it behaves * like {@link mergeAll}. However, when a new inner Observable is emitted, * `switch` unsubscribes from the earlier-emitted inner Observable and * subscribes to the new inner Observable and begins emitting items from it. It * continues to behave like this for subsequent inner Observables. * * @example Rerun an interval Observable on every click event * var clicks = Rx.Observable.fromEvent(document, 'click'); * // Each click event is mapped to an Observable that ticks every second * var higherOrder = clicks.map((ev) => Rx.Observable.interval(1000)); * var switched = higherOrder.switch(); * // The outcome is that `switched` is essentially a timer that restarts * // on every click. The interval Observables from older clicks do not merge * // with the current interval Observable. * switched.subscribe(x => console.log(x)); * * @see {@link combineAll} * @see {@link concatAll} * @see {@link exhaust} * @see {@link mergeAll} * @see {@link switchMap} * @see {@link switchMapTo} * @see {@link zipAll} * * @return {Observable} An Observable that emits the items emitted by the * Observable most recently emitted by the source Observable. * @method switch * @name switch * @owner Observable */ export function _switch(this: Observable): T { return this.lift(new SwitchOperator()); } class SwitchOperator implements Operator { call(subscriber: Subscriber, source: any): any { return source.subscribe(new SwitchSubscriber(subscriber)); } } /** * We need this JSDoc comment for affecting ESDoc. * @ignore * @extends {Ignored} */ class SwitchSubscriber extends OuterSubscriber { private active: number = 0; private hasCompleted: boolean = false; innerSubscription: Subscription; constructor(destination: Subscriber) { super(destination); } protected _next(value: T): void { this.unsubscribeInner(); this.active++; this.add(this.innerSubscription = subscribeToResult(this, value)); } protected _complete(): void { this.hasCompleted = true; if (this.active === 0) { this.destination.complete(); } } private unsubscribeInner(): void { this.active = this.active > 0 ? this.active - 1 : 0; const innerSubscription = this.innerSubscription; if (innerSubscription) { innerSubscription.unsubscribe(); this.remove(innerSubscription); } } notifyNext(outerValue: T, innerValue: R, outerIndex: number, innerIndex: number, innerSub: InnerSubscriber): void { this.destination.next(innerValue); } notifyError(err: any): void { this.destination.error(err); } notifyComplete(): void { this.unsubscribeInner(); if (this.hasCompleted && this.active === 0) { this.destination.complete(); } } }