RxJS with React: Actions and Subjects

RxJS and React go together like chocolate and peanut butter: great individually but they become something incredible when put together.

Actions

So in the last article, we looked at how you can use the React built-in hooks useState and useEffect to subscribe to RxJS Observables. We created a custom useObservable hook that we can pass an RxJS Observable to, which will return the current value of the Observable and re-render every time the value changes.

In the React community, there used to be a lot more talk about the concept of Data Down and Actions Up (DDAU). With our useObservable custom hook, we've got the "data down" part cased, so now we need to handle the actions.

What are actions anyway?

Nowadays, there are 2 meanings to what we call React actions:

1. The original meaning, which was just about passing functions (callbacks) as props to communicate with your parent component. There is no 2-way binding in React, so if the child component wants to tell its parent component someone clicked a button or changed an input, it calls a function that was passed to it as a prop, which is sometimes known as the delegate pattern. Actions are just the act of calling those delegate prop functions (callbacks), like onChange or onPause.

2. Redux popularized a new meaning for actions though. In Redux, actions are serializable objects that describe the interaction or intent of the user. Conventionally they have a type property, and optional payload and meta properties as well. The action objects are dispatched into the Redux store by calling the store's dispatch() method, and passing the action object as an argument.

But whether actions are the act of calling a delegate callback function, or an object describing the intent passed to a dispatch call, it still represents the idea that "something happened", and I would like to send a message to my parent component, the store, or whatever, describing what happened.

Subjects

RxJS Subjects are a great fit for dispatching actions. Subjects are special Observables that can also act as observers, because they implement the observer interface, which is just a fancy way of saying it has the methods next(), error(), and complete(). You can dispatch actions by calling a subject's .next() method, and passing whatever description of the event you need as an argument: Whatever argument we pass to .next() is emitted to all the subject's observers, sometimes called subscribers.

We can use RxJS's Subject to implement something like a Redux store. We can derive our State observable, as a combination of the current state, and an observable of actions which we get from using our Action Subject.

To get a clearer picture of what that means, let's use the simple useObservable custom hook we wrote in Part 1 to create a simple count widget.

We'll create an observable of state (count), out the observable of actions + the current state:

// this will be an observable of `increment` or `decrement` strings
const action$ = new Subject();
// map the action strings to a state update number
const update$ = action$.pipe(
  map((action) => (action === "increment" ? +1 : -1))
);
// update the state by summing the state and the update
const count$ = update$.pipe(
  startWith(0), // our initial state will be 0
  scan((count, update) => count + update)
);

...and a widget component that uses the count$ observable and the custom useObservable hook to get a count state, and also uses the action$ Subject to update the state by passing increment or decrement actions to the action$.next() method.

const CountWidget = () => {
  const count = useObservable(count$);
  return (
    <div className="count-widget">
      <button onClick={() => action$.next("decrement")}>-</button>
      <span>{count}</span>
      <button onClick={() => action$.next("increment")}>+</button>
    </div>
  );
};

Here is the simple demo of the above.

This is simplistic but the idea can be expanded to something more useful. If we combine the same technique with our User Name Fetcher from Part 1, we could easily add a paging feature to give the user the ability to navigate through a list of items. We create a callback function that takes an Observable of the 'back' and 'forward' actions dispatched in our component, and based on that, it fetches new "pages" of users, by increasing or decreasing the page query parameter in the API call.

Though the example is a little more involved, the idea is the same, create an observable of "page number" which is derived from the actions, and use the page$ observable to derive the list of names from an API call.

Something like useReducer

One of the nice aspects of the React built-in hook useReducer is that you can define the reducer outside of the component. You can test the reducer function independently, and you know when you pass it to useReducer React will just update the state and re-render the component automatically.

Let's change our useObservable hook to have the same qualities.

To achieve this, we will alter our useObservable hook to take a function instead. The function passed to useObservable will receive an Observable of actions (the actions we dispatch from the component) as an argument, and will be expected to return an Observable of our new state. We’ll model the API for our custom hook afteruseReducer(), so it will return a tuple of

[state, dispatch].

This way, we can leave it up to the developer how they want to respond to the dispatched actions and how it will affect the state.

Something like this:

useObservable((action$) => {
  // let the developer decide how the action$ Observable affects the state
  actions$.pipe(/* … */);
  // returns an observable that emits the new state
  return newState$;
});

So to implement our new useObservable() custom hook we will:

• take a callback function fn as an argument;
• create an RxJS Subject as our action$ observable;
• create a dispatch function that passes it's argument to action.next();
• create a state$ Observable by calling the fn callback and passing the action$ as an argument
• pull the state out of the state$ observable using the same useState/useEffect technique as before
• return the new state and the dispatch function as a [state, dispatch] tuple

With that we end up with something like this:

const useObservable = (callback) => {
  // create the action$ observable only 1 time
  const action$ = useRef(new Subject()).current;
  // the dipatch function is memoized with useCallback()
  const dispatch = useCallback((v) => action$.next(v), [action$]);
  // store the callback on a ref, ignoring any new callback values
  const fn = useRef(callback).current;

  const [state, setState] = useState();
  useEffect(() => {
    // use the callback to create the new state$ observable
    const state$ = fn(action$);

    const sub = state$.subscribe(setState);
    return () => sub.unsubscribe();
  }, [fn, action$]);

  return [state, dispatch];
};

This looks a little like useReducer now, except that while useReducer is limited to synchronous updates to state, our useObservable can update state over time. Also, our useObservable is a safe async-hook, because it unsubscribes on clean-up, so you don’t have to worry about updating a components state after it has been unmounted.

Updating the example

Now, with that in place, we can define a getUserNames() function that follows our expected useObservable interface. Our getUserNames() function can be separate, isolated from our component. We can test it independently and, in theory, use the same functionality in different components. We'll extract the name fetching functionality into its own file and export the function getUserNames.

import { map, startWith, scan, switchMap } from "rxjs/operators";
import { ajax } from "rxjs/ajax";

const api = `https://randomuser.me/api/?results=5&seed=rx-react&nat=us&inc=name&noinfo`;
const getName = (user) => `${user.name.first} ${user.name.last}`;

export const getUserNames = (action$) => {
  const actionMap = {
    forward: +1,
    back: -1,
  };

  const page$ = action$.pipe(
    scan((page, action) => page + actionMap[action], 1),
    startWith(1)
  );

  return page$.pipe(
    switchMap((page) => ajax.getJSON(`${api}&page=${page}`)),
    map(({ results }) => results.map(getName))
  );
};

Then our component would import getUserNames and along with our new useObservable and look something like this:

function App() {
  const [names, dispatch] = useObservable(getUserNames);

  return (
    <div className="App">
      <h1>RxJS with React</h1>
      <List items={names} />
      <button onClick={() => dispatch("back")}>⇦</button>
      <button onClick={() => dispatch("forward")}>⇨</button>
    </div>
  );
}

Here is the example in full:

I think this is a really nice pattern: it is obvious what the component does, the presentation is decoupled from how the data is actually retrieved, it follows the flux pattern, and generally aligns nicely with React model of component state and side effects.

---

This is really just scratching the surface though, our useObservable hook could be improved in many ways, including exposing the current state to the callback function, using memoization and other techniques to improve the performance, and offering some way to allow component props/state to be exposed to the callback function as a stream of props.

If you'd like to see a more robust implementation of these ideas, you can check out my use-epic library on GitHub which follows a very similar pattern.

BigAB / use-epic

Use RxJS Epics as state management for your React Components

There is so much more that could be done when mixing React with RxJS: Animations, real-time updates, entity stores... the list goes on and on. If you'd be interested in any of those topics, let me know in the comments.

Comments

[Farzad]

A very well written article, thank you for your time and effort :)

I am amazed at how effortless and easier is over something like Redux + RxJS.

Your useObservable() custom hook is much easier than useReducer(). I have added a tiny tweak on top of your useObservable, 1) changed the return from an array to object, 2) returning action$ alongside state and dispatch, so if there was a need for other observables to be based on initial observable, you could easily pass the action$ to those ones.

Here is a little demo I have put together, hope it helps those interested to learn RxJS :)

codesandbox.io/s/falling-cloud-1m4v8

Thanks

[Robert Grant]

This is really great. Thank you. Two small things:

1) the paging example only works for me if I go to the example site directly
2) I didn't really understand the use of scan's default of 1 in the getUserNames example, where you also had to use startWith(1) - I assume it's because you need a value even before the first event is emitted, but it took a little research of my own to hit that.

Thank you again!

Rob

[Adam L Barrett]

Yeah that’s a bit of a hack but, I wanted the state to be there (1) immediately on subscribing so i needed the startsWith(1), but the scan will wait for 2 events if you don’t provide a default value, so by providing it and the startWith it works