RxBus.js

function RxBus() {

  function remove(xs, x) {
     xs.splice(xs.indexOf(x), 1);
  }

  function innerSubscription(observable) {
    var disposable;

    function cancel() { remove(subscriptions, subscription); }
    
    function push(message) { rxBus.push(message); }
    
    function start() {
      disposable = observable.subscribe(push, cancel);
    }

    function stop() {
      disposable && disposable.dispose();
    }

    var subscription = {
      start : start, stop : stop
    };

    subscriptions.push(subscription);

    observers.length > 0 && start();  
    
    return subscription;
  }

  var subscriptions = [];
  var observers = [];

  var rxBus = Rx.Observable.create(function(observer) {
      observers.push(observer)
      if (observers.length == 1) {
        subscriptions.forEach(function(subscription) { subscription.start(); })
      }
      return function() {
        remove(observers, observer);
        if (observers.length === 0) {
          subscriptions.forEach(function(subscription) { subscription.stop(); })
        }
      }
  });

  rxBus.ofType = function(messageType) { 
    return rxBus.where(function(message) { return message.message == messageType; });
  };

  rxBus.push = function (message) {
    observers.map(identity).forEach(function(observer) {
      observer.onNext(message)
    });
    return rxBus;
  };

  rxBus.plug = function (observable) {
    innerSubscription(observable);
    return rxBus;
  };

  return rxBus;
}