Observable with Lift and Mutate

Observable.js

var Subscriber = require('./Subscriber'),
    noop = require('./support/noop')
    ;

module.exports = (function Observable() {
    
    function fixDisposable(subscriber, disposable) {
        var disposableType;
        if(disposable != null) {
            if((disposableType = typeof disposable) === 'function') {
                return subscriber.add({ dispose: disposable });
            } else if(disposableType === 'object') {
                return subscriber.add(disposable);
            }
        }
        return subscriber;
    }
    
    function Observable(subscribe, mutate, transform) {
        this._subscribe = subscribe;
        this._mutate = mutate;
        this._transform = transform;
    }
    
    var create = Observable.create = function(subscribe, mutate, transform) {
        return new Observable(subscribe, mutate, transform);
    }
    
    var subscribe = Observable.prototype.subscribe = function(onN, onE, onC, onD) {
        var subscriber = (onN != null && typeof onN === 'object' ?
                onN :
                Object.create(Subscriber.prototype, {
                    _onNext:      { value: typeof onN === 'function' ? onN : noop },
                    _onError:     { value: typeof onE === 'function' ? onE : noop },
                    _onCompleted: { value: typeof onC === 'function' ? onC : noop },
                    _onDisposed:  { value: typeof onD === 'function' ? onD : noop },
                    disposed:     { value: false }
                })
            ),
            mutate = this._mutate,
            transform = this._transform
            ;
        
        if(transform) {
            subscriber = transform(subscriber);
        }
        
        return fixDisposable(
            subscriber,
            mutate ?
                mutate(this).subscribe(subscriber) :
                this._subscribe(subscriber)
            );
    }
    
    Observable.prototype.forEach = subscribe;
    
    Observable.prototype.lift = function(transform) {
        return create(this._subscribe, null, transform);
    }
    
    Observable.prototype.mutate = function(mutate) {
        return create(this._subscribe, mutate, null);
    }
    
    return Observable;
}());

usage.js

var Observable = require('./Observable'),
    Subscriber = require('./Subscriber');

var values = Observable.create(function(subscriber) {
  subscriber.onNext({value: 1});
  subscriber.onNext({value: 2});
  subscriber.onNext({value: 3});
  subscriber.onCompleted();
});

values.subscribe(function(value) {
  console.log(JSON.stringify(value));
});

var ints1 = values.map(function(x) {
  return x.value;
});

ints1.subscribe(function(i) {
  console.log(i);
});

var ints2 = values.mutate(function(values) {
  var subscribe = values.subscribe;
  values.subscribe = function(subscriber) {
    return subscribe(function(x) {
      subscriber.onNext(x.value);
    }, subscriber.onError.bind(subscriber), subscriber.onCompleted.bind(subscriber));
  }
  return values;
  /* or
  return values.map(function(x) {
    return x.value;
  });
  */
});

ints2.subscribe(function(i) {
  console.log(i);
});

var ints3 = values.lift(function(subscriber) {
  var onNext = subscriber.onNext;
  subscriber.onNext = function(x) {
    onNext(x.value);
  }
  return subscriber;
  /* or
  return new Subscriber(function(x) {
    subscriber.onNext(x.value);
  }, subscriber.onError, subscriber.onCompleted, subscriber);
  */
});

ints3.subscribe(function(i) {
  console.log(i);
});