Posted on Apr 26, 2021

RxJS pipe as a React hook

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> }