ObservePolyfill.js

export default {
  getInitialState() {
    const data = {};

    this.subscribe(this.props, this.context, (key, value) => {
      data[key] = value;
    });
    this.unsubscribe();

    return { data };
  },

  componentWillMount() {
    this.subscribe(this.props, this.context, this.setData);
  },

  componentWillReceiveProps(props, context) {
    this.subscribe(props, context, this.setData);
  },

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

  setData(key, value) {
    this.setState({
      data: {...this.state.data, [key]: value }
    });
  },

  subscribe(props, context, onNext) {
    const newObservables = this.observe(props, context);
    const newSubscriptions = {};

    for (let key in newObservables) {
      newSubscriptions[key] = newObservables[key].subscribe({
        onNext: (value) => onNext(key, value),
        onError: () => {},
        onCompleted: () => {}
      });
    }

    this.unsubscribe();
    this.subscriptions = newSubscriptions;
  },

  unsubscribe() {
    for (let key in this.subscriptions) {
      if (this.subscriptions.hasOwnProperty(key)) {
        this.subscriptions[key].dispose();
      }
    }

    this.subscriptions = {};
  }
}

This differs from the proposed API in several ways:

• You can't use this.props from observe function, use props parameter instead. The second parameter is context.
• Only calls observe on prop changes, not on state changes.
• Instead of this.data, data is available on this.state.data.

Usage:

var Component = React.createClass({
  mixins: [ObservePolyfill],

  observe(props) {
    return {
      user: UserStore.observeUser(props.userId)
    };
  },

  render() {
    return <div>{this.state.data.user && this.state.data.user.name}</div>
  }
})

assuming observeUser returns an Rx-like Observable.

An Observable is just

observable: {
  subscribe: (observer: {
    onNext: (value) -> (),
    onError: (err) -> (),
    onCompleted: () -> ()
  }) -> (subscription: {
    dispose: () -> ()
  }))
}

It is great for encapsulating async sequence of values (e.g. Flux store changes). Think Promise, but a sequence.

bumped into a bug. instead of

this.setState({
  data: {...this.state.data, [key]: value }
});

it should be

this.setState(function (prevState) {
  return {
    data: {...prevState.data, [key]: value}
  };
});

else you miss out on when observables update synchronously, as is the case in flux.

cheers

Found a small bug/implementation detail. When a component unsubscribes, the old data is not purged.
This is easy to fix by adding this code at the end of the subscribe method:

Object.keys(this.subscriptions)
      .filter(key => !newSubscriptions[key])
      .forEach(key => this.setData(key, undefined))

this.subscriptions = newSubscriptions

This will purge the data for keys that no longer have a subscription.