Rx Observable from Meteor reactive source

rx_meteor.js

import Rx from 'rx';

function fromTrackerSource(source) {
  return fromAutorun(function(observer) {
    try {
      observer.onNext(source());
    } catch (e) {
      observer.onError(e);
    }
  });
}

function fromMeteorSubscription(...subArgs) {
  return fromAutorun(function(observer) {
    const sub = Meteor.subscribe(...subArgs);

    Tracker.autorun(function() {
      try {
        observer.onNext(sub.ready());
      } catch (e) {
        observer.onError(e);
      }
    });
  });
}

function fromAutorun(task) {
  return Rx.Observable.create(function(observer) {
    const comp = Tracker.autorun(function() {
      task(observer);
    });

    return function() {
      comp.stop();
    };
  });
}

function makeHot(def, o) {
  return o.multicast(new Rx.BehaviorSubject(def)).refCount();
}

Rx.Observable.fromTrackerSource = function(source) {
  return makeHot(null, fromTrackerSource(source));
};

Rx.Observable.fromMeteorSubscription = function(...subargs) {
  return makeHot(false, fromMeteorSubscription(...subargs));
};

export default Rx;

// example usage in React
// 
//   componentWillMount() {
//     this.issueSub$ = Rx.Observable.fromMeteorSubscription('Issues.one', this.props._id);
//     this.issue$ = Rx.Observable.fromTrackerSource(() => Issues.findOne(this.props._id));
// 
//     this.subs = [
//       this.issue$.subscribe(issue => this.setState({issue})),
//       this.issueSub$.subscribe(ready => this.setState({ready})),
//     ];
//   },
// 
//   componentWillUnmount() {
//     this.subs.forEach((s) => s.dispose());
//   },