RxJS pipe as a React hook

#react #rxjs #javascript #typescript

Ever tried to use an Rx Observable in React? Then you know what's the problem with this code:

function App() {
  let [time, setTime] = useState(0);
  timer(0, 1000)
    .pipe(
      filter(x => x % 2),
      map(x => x + '!')
    )
    .subscribe(setTime);

  return <h1>{ time }</h1>
}

Yeah, it subscribes to the timer with each render. Which triggers setTime on each timer emission. Which leads to a re-render. Which leads to... well, you know, memory leaks and weird behaviour. And on top of that it even won't be destroyed with the component unmount.

In this short post I want to share with you a non-canonical idea (probably not original one) how to fix that!

tl;dr: online playground with hook pipe

🪝 The hook

We could devise a custom react hook, that will fix that. Lets use a useEffect hook, which will subscribe to the source, and push messages to our observer (setTime in the example above)

let useObservable = (observable, observer) => {
  // useEffect with empty deps will call this only once
  useEffect(() => {
    let sub = observable.subscribe(observer); // connect
    return () => sub.unsubscribe(); // < unsub on unmount
  }, []);
}

And it will be used like this:

function App() {
  let [time, setTime] = useState(0);
  useObservable(
    timer(0, 1000)
      .pipe(
        filter(x => x % 2),
        map(x => x + '!')
      ),
    setTime
  );

  return <h1>{ time }</h1>
}

Which looks react-ish... but not rx-ish.
Not nice 😕. We can do better!

So let's explore another way!

🗞️ RxJS pipes

But before we continue, a quick reminder of RxJS pipe operator mechanics:

Roughly speaking RxJS pipe operator (like, map) is just a function that takes one Observable and returns a new Observable:

(source: Observable<A>) => Observable<B>

So when we subscribe to the resulting Observable<B>, operator subscribes to the source Observable<A>. And when that source emits a value, operator applies its logic to it (map, filter, etc) and decides what, when, and how to push to the resulting Observable<B>. map will push modified values, filter will push only values that satisfy given condition.

Okay, back to hooks

🪝🗞️ The hook pipe

We can modify the hook to implement the Rx Operator interface, while still enclosing a useEffect hook.

Let's start with how we'll use it in a component:

function App() {
  let [time, setTime] = useState(0);

  timer(0, 1000)
    .pipe(
      filter(x => x % 2),
      map(x => x + '!'),
      useUntilUnmount()
    )
    .subscribe(setTime);

  return <h1>{ time }</h1>
}

And here's it's implementation:

function useUntilUnmount() {
  // Observable => Observable interface
  return source => new Observable(observer => {
    // create a new Subscription
    // we'll use it to handle un-mounts and unsubscriptions
    let sub = new Subscription();

    // this is run only once
    useEffect(() => {
      // connect observer to source
      sub.add(source.subscribe(observer));
      // on unmount -- destroy this subscription
      return () => sub.unsubscribe();
    }, []);

    // return sub to handle un-subscriptions
    return sub;
  });
}

This is really just 8 lines of code.

Disclaimer: while being leak-free and working as promised, this might not be the best way to use Observables in React. Tried <$> fragment already?

🛸💨 Outro

Try our shiny hook pipe (with dependencies!) in this online playground and leave a comment here with your opinion!

And in the future, when pipeline operator |> lands in JS, we'll probably substitute the subscribe with our custom hook subscribe. Like this:

function App() {
  let [time, setTime] = useState(0);

  timer(0, 1000)
    |> filter(x => x % 2)
    |> map(x => x + '!')
    |> subscribeHook(setTime)

  return <h1>{ time }</h1>
}