Function Subscription

src/Observable.js:75–126

(observer, subscriber)

}

function Subscription(observer, subscriber) {
    // Assert: subscriber is callable
    // The observer must be an object
    this._cleanup = undefined;
    this._observer = observer;

    // If the observer has a start method, call it with the subscription object
    try {
        let start = getMethod(observer, "start");

        if (start) {
            start.call(observer, this);
        }
    }
    catch(e) {
        // HostReportErrors(e);
    }

    // If the observer has unsubscribed from the start method, exit
    if (subscriptionClosed(this))
        return;

    observer = new SubscriptionObserver(this);

    try {

        // Call the subscriber function
        let cleanup = subscriber.call(undefined, observer);

        // The return value must be undefined, null, a subscription object, or a function
        if (cleanup != null) {
            if (typeof cleanup.unsubscribe === "function")
                cleanup = cleanupFromSubscription(cleanup);
            else if (typeof cleanup !== "function")
                throw new TypeError(cleanup + " is not a function");

            this._cleanup = cleanup;
        }

    } catch (e) {

        // If an error occurs during startup, then send the error
        // to the observer.
        observer.error(e);
        return;
    }

    // If the stream is already finished, then perform cleanup
    if (subscriptionClosed(this)) {
        cleanupSubscription(this);
    }
}

Subscription.prototype = nonEnum({
    get closed() { return subscriptionClosed(this) },