Subject implementation

Subject.js

// Both Subject and ReplaySubject

function Subject() {
    if (!(this instanceof Subject)) {
        throw new TypeError('"Subject" cannot be called as a function. Use the "new" keyword to construct a new "Subject" object');
    }

    var _this = this;

    ['unsubscribe', 'next', 'error', 'complete']
        .forEach(function (prop) {
            Object.defineProperty(_this, prop, {
                value: _tryMethod.bind(_this, prop)
            });
        });

    return Object.defineProperties(_this, {
        _id: {
            writable: true,
            value: 0
        },
        _subscribers: {
            value: []
        },
        closed: {
            get: Object.hasOwnProperty.bind(_this, '_hasClosed')
        },
        subscribe: {
            value: function (onNext, onError, onComplete) {
                // The first argument can either be an Observer interface or an "onNext" function
                var subscriber = typeof onNext !== 'function' && Object(onNext) === onNext ? onNext : {
                    next: onNext,
                    error: onError,
                    complete: onComplete
                }

                var id = _this._id++;
                _this._subscribers[id] = subscriber;

                return Object.defineProperties({}, {
                    closed: {
                        get: function () {
                            return _this._subscribers[id] === undefined;
                        }
                    },
                    unsubscribe: {
                        value: function () {
                            // Clear the subscriber object from the internal array
                            _this._subscribers[id] = undefined;
                        }
                    }
                });
            }
        }
    });
}

// "this" is bound to the subject object
function _tryMethod(method, value) {
    if (!this.closed) {
        if (method !== 'unsubscribe') {
            this._subscribers.forEach(function (subscriber) {
                try {
                    subscriber[method].apply(subscriber, method !== 'complete' ? [value] : []);
                } catch (ex) {
                    // Ignore
                }
            });
        } else {
            // Ensure the "_hasClosed" property cannot be mutated anymore
            Object.defineProperty(this, '_hasClosed', {});
            this._subscribers.length = 0;
        }

        if (method === 'error' || method === 'complete') {
            this.unsubscribe();
        }
    }
}

// Subject Example(s)

var subject = new Subject();
var s1 = subject
    .subscribe({
        next: function (value) {
            console.log('#1::next::', value);
        },
        error: function (err) {
            console.log('#1::error::', err);
        },
        complete: function () {
            console.log('#1::complete::');
        }
    });

var s2 = subject
    .subscribe({
        next: function (value) {
            console.log('#2::next::', value);
        },
        error: function (err) {
            console.log('#2::error::', err);
        },
        complete: function () {
            console.log('#2::complete::');
        }
    });

subject.next(100);
subject.next(101);

s1.unsubscribe();

subject.next(102);
s2.unsubscribe();

subject.next(103);
subject.complete();
subject.next(104);