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());
//   },

Hi, thanks for sharing this gist. I have problem though with fromMeteorSubscription.

In the code below commented out items$ Observable from array works fine, but fromMeteorSubscriptoin doesn't.

Am I missing something?

// client/main.js
import Rx from './rx_meteor';
import { run } from '@cycle/core';
import { div, makeDOMDriver } from '@cycle/dom';

Meteor.startup(() => {

    function main() {

        // Works
        // const items$ = Rx.Observable.just([
        //  {name: 'hello'},
        //  {name: 'foo'},
        //  {name: 'bar'}]);

        // Doesn't work
        const items$ = Rx.Observable
            .fromMeteorSubscription('items');

        const vtree$ = items$.map(items => 
            div('.list',
                items.map(item =>
                    div('.item', item.name))))

        return {
            DOM: vtree$
        }
    }
    run(main, { DOM: makeDOMDriver('#app') });
});