function ObservableSubscription (stream, observer) {
    var self = this;

    // Set attributes
    this._source = stream.fork();
    this.closed = false;

    // Don't let users subscribe to an already completed stream
    if (stream.ended) {
        if (observer.error) {
            observer.error(new Error('Subscribe called on an already completed stream.'));
        }

        this._cleanup();

        return;
    }

    // Consume the stream and emit data to the observer
    this._source = this._source.consume(function (err, x, push, next) {
        if (err) {
            push(null, nil);
            if (observer.error) {
                observer.error(err);
            }
            self._cleanup();
        }
        else if (x === nil) {
            if (observer.complete) {
                observer.complete();
            }
            self._cleanup();
        }
        else {
            if (observer.next) {
                observer.next(x);
            }
            next();
        }
    });

    this._source.resume();
}
An implementation of the TC39 Subscription object
https://tc39.github.io/proposal-observable/#subscription-objects

This class is intended for internal use only.

Constructor takes a source highland stream, and an observer object with
an optional next, error, and complete methods.

Returns a subscription object with a closed boolean and unsubscribe
method.

@id ObservableSubscription
@name ObservableSubscription
@param {stream} stream - Highland stream to subscribe to
@param {object} observer - Observer to publish from stream subscription
@api private