React, redux and rxjs example

connectRx.jsx

import React from 'react';
import PropTypes from 'prop-types';
import { Observable } from 'rxjs';
import shallowEqual from 'shallowequal';
import RxComponent from './RxComponent';

const connectRx = handleStream => ComposedComponent => class ConnectRx extends RxComponent {
  static contextTypes = {
    store: PropTypes.object.isRequired,
  };
  engine() {
    const { store } = this.context;
    const state$ = Observable
      .create((observer) => {
        observer.next(store.getState());
        return store.subscribe(() => observer.next(store.getState()));
      })
      .publishReplay(1)
      .refCount();
    return handleStream(
      state$,
      this.props$,
      action => store.dispatch(action),
    )
      .distinctUntilChanged(shallowEqual)
      .map(props => <ComposedComponent {...props} />);
  }
};

export default connectRx;

createEventStream.js

import { Subject } from 'rxjs';

const createEventStream = () => {
  const subject = new Subject();
  return {
    stream$: subject.map(e => e),
    handler: e => subject.next(e),
  };
};

export default createEventStream;

example.js

import createEventStream from './createEventStream';
import connectRx from './connectRx';

const SmartComponent = connectRx((state$, props$, dispatch) => {
  const { stream$: streamSubmit$, handler: onSubmit } = createEventStream();
  const router$ = props$.pluck('router');
  return streamSubmit$
    .switchMap((e) => Observable
      .from(dispatch(someSyncAction()))
      .concat(router$.first().do(router => router.push('/some/url')))
    )
    .ignoreElements()
    .merge(state$)
    .combineLatest(props$, (state, ownProps) => ({
      name: state.user[ownProps.userId],
      onSubmit
    }))
})(DumbComponent);

RxComponent.js

import { Component, Children } from 'react';
import { ReplaySubject, Subject } from 'rxjs';

class RxComponent extends Component {
  state = {
    component: null,
  };
  componentWillMount() {
    this.props$.next(this.props);
    this.subscription = this.clientMounted$
      .ignoreElements()
      .merge(this.engine())
      .subscribe(
        component => this.setState({ error: null, component }),
        error => this.setState({ error }),
      );
  }
  componentDidMount() {
    this.didMountSubject.next(true);
  }
  componentWillReceiveProps(props) {
    this.props$.next(props);
  }
  shouldComponentUpdate(nextProps, nextState) {
    return this.state.component !== nextState.component;
  }
  componentWillUnmount() {
    this.subscription.unsubscribe();
  }
  engine() { // eslint-disable-line
    throw new Error('Should be implemented');
  }
  props$ = new ReplaySubject(1);
  didMountSubject = new Subject();
  clientMounted$ = this.didMountSubject
    .startWith(false)
    .publishReplay(1)
    .refCount();
  render() {
    const { component, error } = this.state;
    if (error) throw error;
    return component ? Children.only(component) : null;
  }
}

export default RxComponent;

I been having some problems with timing/teardown issues for react-redux. If you been using for instance redux-thunk or redux-api-middleware or just doing some async actions with react-redux I think you can benefit from my code above.

Here is an example of problematic code with the current react-redux. You do some async action and when it resolves you want to redirect to another path. This is of course doable with redux-api-middleware but what happens if your component unmounts? Then you probably want to teardown that behavior.

const SmartForm = connect(
  null,
  (dispatch, { router }) => ({
    onSubmit: () => dispatch(asyncAction()).then(() => router.push('/some/path')),
  }),
)(DumbForm);