RxJS and React: basics

RxJS is a library that gives you robust tools to work with Observables.

What are Observables?

In short, they're a type of value that you can listen and react to, to control or alter the flow of data in your application.

Here's a simple example:

import { map, of, tap } from 'rxjs'

// use of() create an observable from the value 'hello'
const value$ = of('hello').pipe(
  map((value) => {
    return value + ' world'!
  }),
  tap((value) => {
    console.log('pipeline says:', value)
  })
)

value$.subscribe({
  next(value) {
    console.log('subscriber says:', value)
  },
})

Console:

pipeline says: hello world
subscriber says: hello world

Those imports, map, tap, of are called "operators" in RxJS. They are provided to help give you tools to combine, modify or create new observables from things like iterables, fetch requests, promises, constants etc.

Involving React

React provides you tools that let your application react to changes in its state. The classic example is a counter, using useState to give us a reactive value display:

function ReactCounter() {
  const [count, setCount] = useState(0)

  return (
    <div>
      <p>{count}</p>
      <button onClick={() => setCount(count + 1)}>
        increment
      </button>
    </div>
  )
}

There is one problem to overcome before we can marry RxJS to React. We can't just dump an observable into our component and expect React to respond to changes. For that, we need to create a custom hook. Luckily, it's really easy.

import { useState, useEffect } from 'react'
import { Observable } from 'rxjs'

function useObservable<T>(source$: Observable<T>, initial: T): T {
  const [value, setValue] = useState(initial)

  useEffect(() => {
    const subscription = source$.subscribe((nextValue) => {
      setValue(nextValue)
    })

    return () => subscription.unsubscribe()
  }, [source$])

  return value
}

There are a few things going on here:

1. We need to make React aware of changes in our Observable somehow - useEffect is perfect for this.
2. Our observable, source$, is being passed as a dependency of useEffect. When it mounts, a subscription will be created that listens for changes. When it unmounts, the subscription will be removed.
3. Our subscription accepts a callback function. That function will be invoked with the new value of our observable if it changes.
4. In that callback, we're altering the state of our hook to match the state of our observable. Viola! React is now reacting to our Observable.

A minor detail for now, but it will become very important later. We need to provide an initial value, because this useEffect/useState system is not immediate. The first render will not have a value.

Let's see what this hook looks like in action:

import { BehaviorSubject } from 'rxjs'

// analagous to useState(0)
const count$ = new BehaviorSubject(0)

// analagous to () => setState(count + 1)
function increment() {
  const value = count$.getValue()

  count$.next(value + 1)
}

function RxJSCounter() {
  const count = useObservable(count$, 0)

  return (
    <div>
      <p>{count}</p>
      <button onClick={increment}>increment</button>
    </div>
  )
}

Notice how our application state, count, is being controlled completely outside of the scope of our component?

This is a very powerful distinction from our original Counter. The new one is no longer dependent on React. It's not dependent on any framework, only observables.

useObservable gotcha

That minor detail I mentioned earlier about initial state? Let's go back to that for a moment.

Currently, we can't depend on an Observable's state to be up-to-date in every component that depends on it. The return value of useObservable is always initialState until a subscription is created by useEffect, the callback is triggered, and then state is updated. This happens on a per-component basis, meaning a parent component can't guard a child component from this behavior.

Let me give you an example. First, let's create our state:

// Our observable, either a user object or null 
const $user = new BehaviorSubject<null | { name: string }>(null)

// Mimic something like a network req. 
// Set a user after 2 seconds.
setTimeout(() => {
  $user.next({ name: 'John Smith'})
}, 2000)

// A hook that will only ever return the current user state.
// If it's called and the user state isn't ready, an error
// will be thrown. This might be useful if you're in an area
// of your app where you KNOW the user is already present,
// and you don't want to do a ton of if (!user) checks.
function useUser() {
  const user = useObservable($user, null)
  if (!user) {
    throw new Error('user not initialized')
  }
  return user 
}

Now, our components:

function Outer() {
  const user = useObservable($user, null)
  if (!user) return null 

  return <Inner />
}

function Inner() {
  const user = useUser()

  return (
    <div>
      <p>hello, {user.name}</p>
    </div>
  )
}

This feels intuitive, right? The Outer component is ensuring a User exists before rendering the Inner component. The Inner component is calling useUser, because a User surely exists in state somewhere.

This results in an error.

Why? Because useUser() eventually calls useObservable(), which does this:

const [value, setValue] = useState(initial)
...
return value

See the issue? The initial state is always returned on first render.

There are a couple of ways around this.

Initial state solution 1 - Context

The first way to get around this issue is by accessing values in hooks/components through a context. The context is created with the initial state (null, in our case) and will be updated with the value of the $user observable, as returned by the useObservable hook.

// Our initial state 
const initial = null 

// Create a react context with the initial state 
const UserContext = createContext(initial)

// Create a provider which provides the observable value
// You'll have to place this somewhere in your application
const UserProvider = ({ children }) => {
  const user = useObservable($user, initial)

  return (
    <UserContext.Provider value={user}>
      {children}
    </UserContext.Provider>
  )
}

function useUser() {
  const user = useContext(UserContext)
  if (!user) {
    throw new Error('user not initialized')
  }
  return user 
}

This works because all attempts at accessing the Observable's value are funneled through a single value - the Context. When it updates, they all update.

Personally, I don't like this solution very much. It's verbose, and you're going to have to create lots of contexts unless you do something like creating a fat context that injects all of the observable values you care about.

Initial state solution 2 - Cache

Another solution is to create a key:value store, where the key is the observable, and the value is the last emitted state.

When the useObservable hook is called by a component that just mounted, it will first look at the cache to synchronously inject the last emitted value as the initial state.

const cache = new Map<Observable<any>, any>()

function useObservable<T>(source$: Observable<T>, initial?: T) {
  /**
   * Use the cache for the observable to inject
   * the initial state synchronously.
   */
  const hasCached = cache.has(source$)
  if (!hasCached) {
    cache.set(source$, initial)
  }
  const cached = cache.get(source$) as T

  /**
   * Make React aware of changes to the observable
   */
  const [value, setValue] = useState(cached)
  useEffect(() => {
    const subscription = source$.subscribe((nextValue) => {
      cache.set(source$, nextValue)
      setValue(nextValue)
    })

    return () => subscription.unsubscribe()
  }, [source$])

  return value
}

In this manner, you no longer have to worry about the initial state renders that occur between mount and the first real update. You also don't have to worry about creating contexts and rendering providers.

The downside here is that every instance of this hook will be updating the cache when a new event is triggered. You could build in some logic to fix this, but that's a bit beyond this scope.

Initial state solution 3 - Redux, or other

You could also configure your application such that Observables push their state to a library like Redux. Then, instead of your components directly subscribing to Observables, your component selects the state from the Redux store.

As always, there will likely be some boilerplate involved, but it should be significantly less than using purely Redux for both state management and state access.

Summary

😴 That was a lot

In short, RxJS is a robust library that allows you to build applications in terms of the data and what that data is doing. Barring a few React-specific obstacles to work around, the two play well with each other, and it doesn't require much work at all to get up and running.

Thanks for reading 👋