Reactive Extensions for JavaScript, or RxJS, is a library that has a twofold purpose.
It creates an Observable primitive that is either synchronous or asynchronous, and it includes a rich library of functions that can be used to create observables, transform, filter, join, and multicast observables, provides error handling, and more.

If that sounds like a lot - it is.

While RxJS is commonly used in Angular projects due to the fact that it is a peer dependency, it can be overlooked by software engineers building applications using React - or other frontend JavaScript frameworks for that matter.

Let me be clear - you do not need to use RxJS with React.

Promises, the useEffect() hook, and libraries such as Axios provide much of what a typical React application requires for asynchronicity and fetching data.
What RxJS with React does provide is the ability to write pure functions for event streams, effectively handle errors within a stream of data, and easily fetch data using the native Fetch and WebSocket APIs.

In this article, I'd like to share how we use RxJS with React at LiveLoveApp for rapidly developing prototypes and applications for our clients.

Using fromFetch()

One advantage to using RxJS is the provided fromFetch() function that uses the native Fetch API with a cancellable AbortController signal.

Let's look at how you might use Axios for cancellation:

import { get } from "axios";
import { Button } from "@mui/material";
import { useCallback, useEffect, useState } from "react";

export default function App() {
  const [user, setUser] = useState(null);
  const controller = new AbortController();

  useEffect(() => {
    const id = 2;
    get(`https://reqres.in/api/users/${id}`, {
      signal: controller.signal
    }).then((response) => {
      try {
        setUser(response.data.data);
      } catch (e) {
        console.error(`Error fetching user`);
      }
    });
  }, []);

  const handleOnCancel = useCallback(() => {
    controller.abort();
  }, []);

  return <Button onClick={handleOnCancel}>Cancel</Button>;
}

Let's quickly review the code above:

• First, we create a new instance of the AbortController class.
• Then, as a side effect, we use Axios' get() method to fetch a user from the API, providing the AbortController's signal.
• Finally, in the handleOnCancel() callback function we invoke the abort() method on the AbortController instance to cancel the fetch request.

When using RxJS's fromFetch() function it is not necessary to wire up an AbortController signal.
Rather, we can cancel the fetch request by emitting either an error or completion notification.

import { Button } from "@mui/material";
import { useCallback, useEffect, useState } from "react";
import { Subject } from "rxjs";
import { fromFetch } from "rxjs/fetch";
import { concatMap, takeUntil, tap } from "rxjs/operators";

export default function App() {
  const [user, setUser] = useState(null);
  const cancel$ = new Subject();

  useEffect(() => {
    const id = 2;
    const subscription = fromFetch(`https://reqres.in/api/users/${id}`)
      .pipe(
        tap((response) => {
          if (!response.ok) {
            throw new Error(response.statusText);
          }
        }),
        concatMap((response) => response.json()),
        tap(user => setUser(user)),
        takeUntil(cancel$)
      )
      .subscribe();
    return () => subscription.unsubscribe();
  }, []);

  const handleOnCancel = useCallback(() => {
    cancel$.next();
  }, []);

  return <Button onClick={handleOnCancel}>Cancel</Button>;
}

Let's review the code above:

• First, we use the fromFetch() function from RxJS to use the native Fetch API to request a user. This function returns an Observable, that when subscribed to, will initiate the request.
• Within the pipe() method, we first check if the response failed, and if so, we emit an error notification of the response's statusText.
• Next, using the concatMap() operator, we merge the next notification that is emitted from the Observable created internally from the Promise returned from the .json() method.
• Next, we use the takeUntil() operator to notify the outer Observable to complete, and abort the request if necessary, when the cancel$ subject emits a next notification.
• Finally, within the handleOnCancel() callback function we invoke the next() notification on the cancel$ Subject.

The key takeaways are:

• RxJS provides functions for interfacing with the native Fetch and WebSocket APIs using asynchronous Observables.
• The fromFetch() operator uses the AbortController internally and cancels the request if the Observable either completes or an error notification is emitted.

How do I handle subscriptions?

It's best to clean up any subscriptions in our application when using RxJS.
While there are a few different approaches to ensuring an Observable that is subscribed to is completed (or unsubscribed from), one method is to invoke the .unsubscribe() method on the Subscription instance that is returned from the subscribe() function.
The teardown function returned from the useEffect() hook is our opportunity to perform any cleanup from the side effect.

De-bouncing an input stream

In this example, we will manage a search$ Observable stream that is denounced before we invoke the onSearch() callback function that is prop to the component.
While we could simply invoke the onSearch() callback function on each change to the input value, we want to avoid excessive network requests and repaints in the browser.

import CancelIcon from "@mui/icons-material/Cancel";
import SearchIcon from "@mui/icons-material/Search";
import { IconButton } from "@mui/material";
import { useEffect, useMemo, useState } from "react";
import { BehaviorSubject } from "rxjs";
import { debounceTime, tap } from "rxjs/operators";

export default function Search(props) {
  const { onSearch } = props;
  const [search, setSearch] = useState("");
  const search$ = useMemo(() => new BehaviorSubject(""), []);

  useEffect(() => {
    search$.next(search);
  }, [search]);

  useEffect(() => {
    const subscription = search$
      .pipe(debounceTime(1000), tap(onSearch))
      .subscribe();
    return () => subscription.unsubscribe();
  }, []);

  return (
    <div>
      <input
        type="text"
        placeholder="Search"
        onChange={(event) => setSearch(event.target.value)}
        value={search}
      />
      {search$.value && (
        <IconButton onClick={() => setSearch("")}>
          <CancelIcon />
        </IconButton>
      )}
      {!search$.value && <SearchIcon />}
    </div>
  );
}

Let's review the code above:

• We have defined a search$ BehaviorSubject with an initial seed value of an empty string.
• When the search state changes the next() method is invoked on the search$ subject with the current value.
• We subscribe to the search$ Observable stream and use the debounceTime() operator to debounce the value changes of the search HTMLInputElement. Within the useEffect() hook we return the teardown callback function that will invoke the unsubscribe() method.

This implementation highlights the use of RxJS to manage a stream of data within our application from the onChange event that is caused by the user interacting with a search input.

The useRxEffect() Hook

Finally, I'd like to share a simple hook that LiveLoveApp uses for our React applications that depend on RxJS.
This hook makes it easy to not worry about subscriptions.

Let's take a look.

import { useEffect } from 'react';
import { Observable } from 'rxjs';

export function useRxEffect(factory: () => Observable<any>, deps: any[]) {
  useEffect(() => {
    const subscription = factory().subscribe();
    return () => subscription.unsubscribe();
  }, deps);
}

The useRxEffect() hooks is intentionally similar to the useEffect() hook provided by React.
The hook expects the factory function to return an Observable that is unsubscribed when the effect teardown callback function is invoked.

Here is a snippet of using the useRxEffect() hook based on the previous code:

import CancelIcon from "@mui/icons-material/Cancel";
import SearchIcon from "@mui/icons-material/Search";
import { IconButton } from "@mui/material";
import { useEffect, useMemo, useState } from "react";
import { BehaviorSubject } from "rxjs";
import { debounceTime, tap } from "rxjs/operators";

export default function Search(props) {
  const { onSearch } = props;
  const [search, setSearch] = useState("");
  const search$ = useMemo(() => new BehaviorSubject(""), []);

  useEffect(() => {
    search$.next(search);
  }, [search]);

  useRxEffect(() => {
    return search$.pipe(debounceTime(1000), tap(onSearch));
  }, []);

  return (
    <div>
      <input
        type="text"
        placeholder="Search"
        onChange={(event) => setSearch(event.target.value)}
        value={search}
      />
      {search$.value && (
        <IconButton onClick={() => setSearch("")}>
          <CancelIcon />
        </IconButton>
      )}
      {!search$.value && <SearchIcon />}
    </div>
  );
}

In the example code above, note that we have replaced the useEffect() hook with our custom useRxEffect() hook to manage the subscribing and unsubscribing from the search$ Observable.

Key Takeaways

If you're considering using RxJS in an existing or new React application, here are some key takeaways based on our experience:

1. RxJS is not necessary to build robust React application.
2. RxJS provides a functional programming implementation for building React applications with event streams, asynchronous data, and more.
3. RxJS implements the Observable primitive that is compatible to Promises (but without async/await).
4. RxJS has a rich library of functions for creating Observables, data transformation and multicasting, handling errors, and more.
5. You can think of RxJS as lodash for events.