Higher Order Function to connect React to Rx Streams.

connectStream.js

import _ from 'lodash';
import { Observable } from 'rx';
import React, { PropTypes } from 'react';

/* Connect Rx Stream to React Component
 * - - - - - - - - - - - - - - - - - - -
 * @param {func}
 *
 * @return {func}
 * - @param {ReactComponent}
 * - @return {WrappedReactComponent}
 *
 * Example Usage...
 *
 * const streamFn = ({ interval }) => {
 *   return Observable.interval(interval).timeInterval().take(3);
 * }
 *
 * class IntervalComponent extends Component {
 *   static propTypes = {
 *     interval: PropTypes.number.isRequired
 *   };
 *
 *   render() {
 *    const { interval = 0, value = 0 } = this.props.state;
 *
 *     return (
 *       <div>
 *         <div>Interval: {interval}</div>
 *         <div>Value: {value}</div>
 *       </div>
 *     );
 *   }
 * }
 *
 * const Interval = connectStream(streamFn)(IntervalComponent);
 *
 * <Interval interval={500} />
 *
 * If you pass in a different `interval` value, it will dispose of the stream, and reconnect with a new value.
 * If you have other props, that as they change that don't impact the stream, you can add the `excludeChangedProps` to the WrappedComponent
 * I.E. static excludeChangedProps = ['propToIgnore'];
 */
export function connectStream(observableFn) {
  const streamFn = observableFn;

  return (WrappedComponent) => {
    return React.createClass({
      propTypes: {
        changedProps: PropTypes.array
      },

      getInitialState() {
        return {};
      },

      getDefaultProps() {
        const { propTypes, excludeChangedProps = [] } = WrappedComponent;

        const changedProps = _(propTypes)
                              .map((prop, key) => key )
                              .filter((key) => excludeChangedProps.indexOf(key) === -1)
                              .value();

        return { changedProps };
      },

      componentDidMount() {
        this._streamFn(this.props);
      },

      componentWillReceiveProps(nextProps) {
        const { props } = this;
        const { changedProps } = props;

        if(changedProps.length > 0) {
          const diffProps = _.reduce(nextProps, (result, value, key) => {
            const isDiff = changedProps.indexOf(key) > -1 && !_.isEqual(value, props[key]);
            return isDiff ? result.concat(key) : result;
          }, []);

          if (diffProps.length > 0) {
            this._streamFn(nextProps);
          }
        }
      },

      componentWillUnmount() {
        this._disposeStream();
      },

      _streamFn(props) {
        this._disposeStream();

        this.subscribe = streamFn(props).subscribe(
          (msg) => {
            if(msg !== null) {
              this.setState(Object.assign({}, { isError: false }, msg));
            }
          },
          (err) => { this.setState({ isError: true, err }); }
        );
      },
      
      _disposeStream() {
        if(this.subscribe && this.subscribe.dispose) {
          this.subscribe.dispose();
        }
      },

      render() {
        return <WrappedComponent {...this.props} state={this.state} />;
      }
    });
  };
}