Observable React elements with RxJS

When I just started using RxJS with React, I was subscribing to observables in componentDidMount and disposing subscriptions at componentWillUnmount.

But, soon I realised that it is not fun to do all that subscriptions(that are just updating property in component state) manually, and written mixin for this...

Later I have rewritten it as "high order component" and added possibility to pass also obsarvers that will receive events from component.

But, shortly after, I realised that for server side rendering, I need to do something quite complicated(in order to wait until all data would be loaded, before rendering - actually, I was building an isomorphic application, and still working on it).

I thought - why not just make observable react elements?

And, I realised that it is completely ok, and much better than previous attempt.

My implementation looks following:

function observableObject(observables) {
  var keys = Object.keys(observables);
  var valueObservables = keys.map(key=>observables[key]);

  if (keys.length === 0) {
    return Rx.Observable.return({});
  }

  return Rx.Observable.combineLatest(valueObservables, (...values)=> {
    let res = {};
    for (let i = 0, l = keys.length; i < l; i++) {
      res[keys[i]] = values[i];
    }
    return res;
  })
}
// creates stream of react elements
function observableElement(Element, observables = {}, observers = {}, props = {}) {
  var propsObservable = observableObject(observables);

  var callbacks = {};

  Object.keys(observers).forEach(key=> {
    var observer = observers[key];
    callbacks[key] = (value)=>observer.onNext(value);
  });

  return propsObservable.map((state)=> {
    return (
      <Element {...props} {...state} {...callbacks}/>
    );
  });
}

And main application, changed from:

React.render(
  <ExampleApplication data={observable} handler={observer}/>
  document.getElementById('container')
);

to something like:

observableElement(ExampleApplication, {data:observable}, {handler:observer}})
  .subscribe((element)=>{
    React.render(element, document.getElementById('container'));
  });

And it is pretty awesome:

• in comparison to mixin, react components again are just views
• observables are easy to composse, e.g. - I can pass one element steam to other one, or jusr combine few of them..
• no need to specify default values for each observable for first render - component would be rendered only when all required data would be loaded

And most awesome part - it happens to be isomorphic, for server side in my request handler, I need to do just this:

observableElement(ExampleApplication, {data:observable}, {handler:observer}})
  .first()
  .subscribe((element)=>{
    var html = React.renderToString(element);
    res.render('main', {appHtml: html});
  });

It will automatically wait for all data required for rendering, and after rendering, as expected - it will dispose everything not needed.

Update 1:

It was good, but not as optimal as it can be...

The problem is that everytime when something changes we are updating whole rendering tree, but react is capable to update olny changed subtree, so I decided to add Proxy element that will not be changing and will update nested component when some of obseravles emit new value.

Also I have rewritten observableElement function as a class allowing to extend it adding new properties.

Update 2

Steams of react element are awesome… But - they are not working well with react context, and as figured out - to support it I should create react element only inside render function...

So, I decided to switch from streams of react elements to stream of react components.

Also I have added method on RxComponent allowing to decorate encolosed component(it allows to apply "high order component" to react compoment enclosed in RxCompoent).

function createProxyComponent(Component, observable, initialState) {
  class RxProxy extends React.Component {
    componentWillMount() {
      this.setState(initialState);
    }

    componentDidMount() {
      this.subscribtion = observable.subscribe((state)=> {
        this.setState(state);
      });
    }

    componentWillUnmount() {
      this.subscribtion.dispose();
    }

    render() {
      return (<Component {...this.props} {...this.state}/>);
    }
  }
  return RxProxy;
}

/**
 * Creates observable form ready to render ReactElements.
 * The same ReactElement would be emitted on every observables combination.
 */
class RxComponent extends AnonymousObservable {
  /**
   * @param {React.Component} Component
   * @param {Object.<string, Rx.Observable>=} observables
   * @param {Object.<string, Rx.Observer>=}observers
   * @param {Object=} props
   */
  constructor(Component, observables = {}, observers = {}, props = {}) {
    super(observer=> {
      const callbacks = {};

      Object.keys(observers).forEach(key=> {
        callbacks[key] = (value)=>observers[key].onNext(value);
      });

      const propsObservable = objectObserver(observables).share();

      const initialState = {};
      const Proxy = createProxyComponent(Component, propsObservable, initialState);
      Proxy.defaultProps = Object.assign({}, props, callbacks);

      return propsObservable
        .do(state=>Object.assign(initialState, state))
        .map(()=>Proxy)
        .subscribe(observer);
    });

    this.params = [Component, observables, observers, props];
  }

  /**
   * Extend defined params
   * @param {Object.<string, Rx.Observable>=} observables
   * @param {Object.<string, Rx.Observer>=} observers
   * @param {Object=} props
   * @returns {RxComponent}
   */
  extend(observables, observers, props) {
    const [Component, prevObservables, prevObservers, prevProps] = this.params;
    return new RxComponent(
      Component,
      observables && Object.assign({}, prevObservables, observables),
      observers && Object.assign({}, prevObservers, observers),
      props && Object.assign({}, prevProps, props)
    );
  }

  /**
   * Extend defined params
   * @param {function(component: React.Component): React.Component} decorator
   * @returns {RxComponent}
   */
  decorate(decorator) {
    const [Component, observables, observers, props] = this.params;
    return new RxComponent(
      decorator(Component),
      observables,
      observers,
      props
    );
  }
}

Now it can be used as following:

let appComponent = new RxComponent(ExampleApplication, {data:observable}, {handler:observer},{property});

// add/replace some properties
// this can be quite useful if you want to have some defaults in your component definition
// or just want to replace something defined before, or add context specific properties

appComponent = appComponent.extend({/* observables */}, {/* observers */}, {/* properties */})

// create new Observalbe with only distinct elements (initialy RxComponent will emit the same element on overy data change)
//
// it is not inside RxComponent to allow side effects implementation when data changes... 
// example of one of possible side effects: srolling to hash after data changes are rendered

appCompomnent = appComponent.distinctUntilChanged();

appComponent
  .subscribe((Component)=>{
    React.render(<Component/>, document.getElementById('container'));
  });