Rewriting 28 RxJS operators as React hooks
Tags: react, rxjs
Published: 2022-06-07 17:37:11

---
title: Rewriting 28 RxJS operators as React hooks
published: true
description: Can React hooks really handle async code well? React hooks for 28 RxJS operators
tags: React, RxJS
cover_image: https://dev-to-uploads.s3.amazonaws.com/uploads/articles/1t8r6fp2t3eegq1za1px.png
series: RxJS in React
---
I wanted to prove that React hooks could do everything RxJS can do, so I tried to convert all 114 operators into React hooks. But there are important differences between React state and RxJS that made me question my goal after 28. 

## Events vs State
RxJS pushes values through streams. If you push `1` into a `new Subject<number>()` several times, everything downstream will receive each value and process it, unless prevented with a `distinctUntilChanged`. 

React reacts to state changes. If you `setState(1)` several times, there is no state change, so nothing will be re-rendered.

To get React to react to multiple events represented by the same value, you need to wrap the value in an event object like `setState({payload: 1})` or pass a DOM event directly. 

Here's an example with increment events:
```tsx
function Incrementer() {
  const [event, setEvent] = useState<MouseEvent>();
  const [count, setCount] = useState(0);

  useEffect(() => {
    if (!event) return;
    setCount(n => n + 1);
  }, [event])

  return <button onClick={setEvent}>Increment: {count}</button>
}
```

Obviously this isn't the best way to implement a simple increment feature, but it shows the most flexible way to deal with values as events as opposed to state. The drawback is it makes React render an extra time when `setCount` is called. 

A simpler (although less flexible) option is `useReducer`:
```tsx
function Incrementer2() {
  const [count, increment] = useReducer(
    (state: number, e: MouseEvent) => state + 1,
    0
  );

  return <button onClick={increment}>Increment: {count}</button>;
}
```

Most async logic occurs before state needs to be set, so this would probably work for the vast majority of cases. 

## Filtering
In RxJS you can filter values to avoid unnecessary downstream calculations.

In React, everything renders, so you need to handle default values.

Hooks cannot be called conditionally, so you need to implement an `enabled` option for each of them, [like React Query does](https://react-query.tanstack.com/guides/disabling-queries), or you need to make them handle default values appropriately. 

## Complexity and Efficiency
There are 114 operators on [the RxJS docs site](https://rxjs.dev/guide/operators). I only converted 28 of them.

When I got to `concatMap` I realized that while it is possible to implement all RxJS operators as hooks, it would take a lot of planning. If there is enough interest, I will do it, but upload it as a GitHub repository, and possibly make an npm package for it.

There were inefficiencies with using React state for some of these operators, because intermediate state needed to be updated synchronously inside `useEffect`s. RxJS seems nicer the more complex an async data flow is.

[React Query](https://react-query.tanstack.com/) is the gold standard for using async data declaratively with hooks. I ended up needing to modify even my simple `useTimer` hook to work more like `useQuery` to take multiple keys in order to work as an inner observable for other operators.

However, even storing all inputs as keys and returning the async data from the latest input is not good enough: I anticipate that a more advanced pattern would be needed to work inside `mergeMap`, where inputs may output many responses in any order, and all of them are necessary.

There may also be more limitations yet to be discovered from the other 86 operators.

## Summary
In short, I stopped because I realized that the complexity required to handle all async pipelines was enough to justify a dedicated npm package, but the value of creating one is marginal since RxJS already exists, is more efficient, and is easy to use in React.

#Notes and Code
(This list is from [the RxJS docs site](https://rxjs.dev/guide/operators))

Check out the demos on [StackBlitz](https://stackblitz.com/edit/react-ts-xuozeg?file=App.tsx).

**Creation Operators**
- [ajax](#-raw-ajax-endraw-)
- [bindCallback](#-raw-bindcallback-endraw-)
- [bindNodeCallback](#-raw-bindnodecallback-endraw-)
- [defer](#-raw-defer-endraw-)
- [empty](#-raw-empty-endraw-)
- [from](#-raw-from-endraw-)
- [fromEvent](#-raw-fromevent-endraw-)
- [fromEventPattern](#-raw-fromeventpattern-endraw-)
- [generate](#-raw-generate-endraw-)
- [interval](#-raw-interval-endraw-)
- [of](#-raw-of-endraw-)
- [range](#-raw-range-endraw-)
- [throwError](#-raw-throwerror-endraw-)
- [timer](#-raw-timer-endraw-)
- [iif](#-raw-iif-endraw-)

**Join Creation Operators**
- [combineLatest](#-raw-combinelatest-endraw-)
- [concat](#-raw-concat-endraw-)
- [forkJoin](#-raw-forkjoin-endraw-)
- [merge](#-raw-merge-endraw-)
- [partition](#-raw-partition-endraw-)
- [race](#-raw-race-endraw-)
- [zip](#-raw-zip-endraw-)

**Transformation Operators**
- [bufferCount](#-raw-buffercount-endraw-)
- [bufferTime](#-raw-buffertime-endraw-)
- [bufferToggle](#-raw-buffertoggle-endraw-)
- [bufferWhen](#-raw-bufferwhen-endraw-)
- [concatMap](#-raw-concatmap-endraw-)

# Creation Operators
## `ajax`

[`fetch`](https://developer.mozilla.org/en-US/docs/Web/API/Fetch_API/Using_Fetch)

## `bindCallback`

See [`bindCallback`](https://medium.com/@cpyle/rxjs-functions-part-1-3275c1cdde21).

You will create a hook creator. Each unique function that takes a callback as its last argument will have its own hook. Here is the hook creator I made:

```tsx
function getUseBindCallback<Args extends any[], Arg>(
  fnWithCb: (...args: [...Args, (arg: Arg) => void]) => void
) {
  return function useBindCallback(initialEvent: Arg, ...args: Args) {
    const [event, setEvent] = useState<Arg>(initialEvent);

    useEffect(() => {
      if (event !== initialEvent) return; // Only run once
      fnWithCb.call(null, ...args, (e: Arg) => {
        setEvent(e);
      });
    }, args);

    return event;
  };
}
```

Here is an example function and corresponding hook:
```tsx
const someFunctionWithCallback = (
  a: string,
  b: number,
  cb: ({ a, b }: { a: string; b: number }) => void
) => {
  setTimeout(() => cb({ a, b }), 3000);
};

const useBindCallback = getUseBindCallback(someFunctionWithCallback);
```

Demo:
```tsx
function UseBindCallbackDemo() {
  const { a, b } = useBindCallback({ a: '-', b: 0 }, 'Callback finished', 1);
  return (
    <div>
      {a} {b}
    </div>
  );
}
```

First in DOM: `- 0`
After 3 seconds: `Callback finished 1`

## `bindNodeCallback`

See [`bindNodeCallback`](https://medium.com/@cpyle/rxjs-functions-part-2-63f06f018fa6).

> It's just like `bindCallback`, but the callback is expected to be of type `callback(error, result)`.

So we can use the same hook creator as with `bindCallback`, but with extra generics `Err`, `Result` and `Arg extends [Err, Result]`. And the callback args will be collected into an array.
```tsx
function getUseBindNodeCallback<
  Args extends any[],
  Err,
  Result,
  Arg extends [Err, Result]
>(fnWithCb: (...args: [...Args, (...arg: Arg) => void]) => void) {
  return function useBindNodeCallback(initialEvent: Arg, ...args: Args) {
    const [event, setEvent] = useState<Arg>(initialEvent);

    useEffect(() => {
      if (event !== initialEvent) return; // Only run once
      fnWithCb.call(null, ...args, (...e: Arg) => {
        setEvent(e);
      });
    }, args);

    return event;
  };
}
```

## `defer`

`defer` relies on the usefulness of lazy subscriptions in RxJS. In React, everything in a component is rendered eagerly, so the way to implement laziness is to change some state from `false` to `true` so some logic downstream can switch from not executing to executing in a render. If it's just computationally expensive, a ternary would work:

```tsx
function DeferDemo() {
  const [enabled, setEnabled] = useState(false);
  const expensiveCalculation = enabled && Math.max(1, 2, 3);

  return (
    <button onClick={() => setEnabled(true)}>
      Enable? Result: {expensiveCalculation}
    </button>
  );
}
```

Sometimes you want to defer a side-effect that occurs when subscribing, like a data fetch. You can create a hook that takes `enabled` as an option and escapes out of a `useEffect` if `!enabled`:

```tsx
export function useData(enabled = true) {
  const [data, setData] = useState<any>();

  useEffect(() => {
    if (!enabled) return;
    fetchData().then(setData);
  }, [enabled]);

  return data;
}
```

## `empty`
I don't think this has a use in React.

## `from`
I don't think this has a use in React. But just in case, this could be the implementation:

```tsx
function useStates<T extends any[]>(states: T) {
  const [stateIdx, setStateIdx] = useState(0);

  useEffect(() => {
    if (stateIdx === states.length - 1) return;
    setStateIdx(stateIdx + 1);
  }, [stateIdx]);

  return states[stateIdx];
}
```

```tsx
function UseStatesDemo() {
  const state = useStates([1, 3, 5, 7, 9]);
  console.log('state', state);
  return <span>UseStatesDemo</span>;
}
```
That demo immediately logs this:
```
state 1
state 3
state 5
state 7
state 9
```

If you care about reacting to events downstream instead of only distinct states, you can modify the hook to wrap each in an event object like `{payload: states[stateIdx]}` or something. 

Most likely you have a scenario where you should just calculate a state using `reduce` on an array. 

## `fromEvent`
This converts a DOM element into a stream of DOM events on that element. You can use `setState` to set event objects to state and react downstream, or you can call `useReducer` in the event handler in the JSX. See "Events vs state" above. 

## `fromEventPattern`
This is similar to `fromEvent`, but it's more general, so it's worth implementing with hooks:

```tsx
type NodeEventHandler = (...args: any[]) => void;

function getUseEventPattern<T>(
  addHandler: (handler: NodeEventHandler) => any,
  removeHandler?: (handler: NodeEventHandler, signal?: any) => void,
  resultSelector?: (...args: any[]) => T
): T {
  const [state, setState] = useState<T>();

  useEffect(() => {
    if (!addHandler || !removeHandler) return;
    const handler = (...e: T[]) => {
      const val = e.length === 1 ? e[0] : e;
      const newState = resultSelector ? resultSelector(val) : val;
      setState(newState as T);
    };
    const retValue = addHandler(handler);
    return () => (removeHandler ? removeHandler(handler, retValue) : undefined);
  }, [addHandler, removeHandler, resultSelector]);

  return state;
}
```

Demo:
```tsx
function addClickHandler<Handler extends EventListener>(handler: Handler) {
  document.addEventListener('click', handler);
}

function removeClickHandler<Handler extends EventListener>(handler: Handler) {
  document.removeEventListener('click', handler);
}

function UseEventPatternDemo() {
  const event = getUseEventPattern<MouseEvent>(
    addClickHandler,
    removeClickHandler
  );

  return <div>{event?.clientX}</div>;
}
```

This listens to document click events and prints out their `clientX` property. 

## `generate`
This is a simple utility that generates a stream of values. It doesn't have unique async logic so I am not going to convert it to hooks. See `from`. 

## `interval`
```tsx
function useInterval(dt = 1000) {
  const [count, setCount] = useState<number>();

  useEffect(() => {
    const interval = setInterval(() => setCount((n = -1) => n + 1), dt);
    return () => clearInterval(interval);
  }, [dt]);

  return count;
}
```

Demo:
```tsx
function UseIntervalDemo() {
  const count = useInterval(2000);
  return <div>{count}</div>;
}
```

## `of`
This immediately returns a value, so you can just define something with `const`. 

## `range`
Another utility method for generating a stream of values. See `from`. 

## `throwError`
I think you can just `throw 'some error'`.

## `timer`
```tsx
function useTimer(wait = 1000, dt?: number) {
  const [count, setCount] = useState<number>();

  useEffect(() => {
    const timeout = count == null && setTimeout(() => setCount(0), wait);
    const interval =
      dt && count != null && setInterval(() => setCount((n) => n + 1), dt);
    return () => {
      if (timeout) clearTimeout(timeout);
      if (interval) clearInterval(interval);
    };
  }, [wait, dt, count]);

  return count;
}
```

Demo:
```tsx
function UseTimerDemo() {
  const count = useTimer(1000, 500);
  return <div>{count}</div>;
}
```

## `iif`
This observable creator could also be written as `const obs$ = defer(() => condition ? observable1 : observable2);`. Refer to `defer`. 

# Join Creation Operators
## `combineLatest`
This reduces input streams into states, so it's very simple in React. 

If you had this in RxJS:
```typescript
const a$ = new BehaviorSubject(1);
const b$ = new BehaviorSubject(2);
const total$ = combineLatest(a$, b$).pipe(
  map(([a, b]) => a + b),
);
```

It would just be this in React:
```tsx
const [a, setA] = useState(1);
const [b, setB] = useState(2);
const total = a + b;
```

## `concat`
Let's convert this to hooks:
```typescript
const result$ = concat(
  timer(2000),
  timer(1000),
  timer(3000),
);
```

If we used 3 `useTimer`s, they would all start at the same time. But with `concat` each observable needs to wait for the previous to complete. So, we need to implement an `enabled` option in whatever observable we want to convert to be part of our `concat` chain. 

So we can modify `useTimer` to take in a 3rd option `enabled = true`, add it to the `useEffect`'s dependency array and add `if (!enabled) return;` at the top of the `useEffect`.

Here is the demo:
```tsx
function ConcatDemo() {
  const count1 = useTimer(2000);
  const count2 = useTimer(1000, null, count1 != null);
  const count3 = useTimer(3000, null, count2 != null);
  const result = [count3, count2, count1].findIndex(count => count != null);
  return <div>{result}</div>;
}
```

First, `count1`, `count2` and `count3` are all `undefined`. Only the first `useTimer` sets a timeout. After 2 seconds, `count1` receives the value of `0` and `count1 != null` evaluates to `true` so the 2nd timeout is kicked off. `result` started at `-1` because all 3 were `undefined` at first, but now it's `2`. After the 2nd timeout finishes, `result` becomes `1`, then after 3 more seconds it ends up as `0`. This is the same behavior as RxJS. 

## `forkJoin`
This operator us overused. It has the same behavior as `combineLatest` for `http` requests (observables that complete after 1st value), but developers just love to flex their RxJS knowledge and use a more limited operator for no reason at all. But occasionally data sources convert to streams of multiple values, and in these cases you actually need `combineLatest`. 

Anyway, if someone provides a concrete example of `forkJoin` that the `combineLatest` example is insufficient for, I'll work on this.

## `merge`
We need several `useEffect`s that react to different inputs but all set the same output state:
```tsx
function useMerge<S1, S2>(s1: S1, s2: S2) {
  const [state, setState] = useState<S1 | S2>();

  useEffect(() => {
    setState(s1);
  }, [s1]);

  useEffect(() => {
    setState(s2);
  }, [s2]);

  return state;
}
```

Demo:
```tsx
function UseMergeDemo() {
  const interval1 = useInterval(900);
  const interval2 = useInterval(1100);
  const interval = useMerge(interval1, interval2);
  return <div>{interval1} {interval2} {interval}</div>
}
```

## `Partition`
Not sure when you'd use this, but it's easy to implement:
```tsx
function usePartition<T>(state: T, partition: (s: T) => boolean) {
  const [state1, setState1] = useState<T>();
  const [state2, setState2] = useState<T>();

  useEffect(() => {
    const setState = partition(state) ? setState1 : setState2;
    setState(state);
  }, [state]);

  return [state1, state2];
}
```

Demo:
```tsx
function UsePartitionDemo() {
  const interval = useInterval(1000);
  const [evens, odds] = usePartition(interval, n => !!(n % 2));
  return <div>{evens} {odds}</div>
}
```

## `race`
For this one, we'll start with two states, each `undefined`, and when the first one gets defined we'll update a third state to that value but then stop paying attention after that.
```tsx
function useRace<S1, S2>(s1: S1, s2: S2) {
  const [state, setState] = useState<S1 | S2>();

  useEffect(() => {
    if (state === undefined && s1 !== undefined) setState(s1);
  }, [state, s1]);

  useEffect(() => {
    if (state === undefined && s2 !== undefined) setState(s2);
  }, [state, s2]);

  return state;
}
```

Demo:
```tsx
function UseRaceDemo() {
  const timer1 = useTimer(3000);
  const timer2 = useTimer(2500);
  const timer2Plus1 = timer2 !== undefined ? timer2 + 1 : timer2;
  const firstToChange = useRace(timer1, timer2Plus1)
  return (
  <div>
    <div>timer1: {timer1}</div>
    <div>timer2Plus1: {timer2Plus1}</div>
    <div>Race: {firstToChange}</div>
  </div>)
}
```

## `zip`
This is another operator I've never had a use for, but it seems fun to convert. 

I don't like the repeated code in my implementation, but I'm not going to spend the time to clean it up. 

```tsx
function useZip<S1, S2>(s1: S1, s2: S2) {
  const [state, setState] = useState<[S1, S2][]>([]);

  useEffect(() => {
    if (s1 === undefined) return;
    setState((oldState) => {
      const missingS1Idx = oldState.findIndex(([oldS1]) => oldS1 === undefined);

      const [oldS1, oldS2] = oldState[missingS1Idx] || [];
      const bothUndefined = oldS2 === undefined;
      return bothUndefined
        ? [...oldState, [s1, undefined]]
        : [[s1, oldS2], ...oldState.slice(2)];
    });
  }, [s1]);

  useEffect(() => {
    if (s2 === undefined) return;
    setState((oldState) => {
      const missingS2Idx = oldState.findIndex(
        ([oldS1, oldS2]) => oldS2 === undefined
      );

      const [oldS1, oldS2] = oldState[missingS2Idx] || [];
      const bothUndefined = oldS1 === undefined;
      return bothUndefined
        ? [...oldState, [undefined, s2]]
        : [[oldS1, s2], ...oldState.slice(2)];
    });
  }, [s2]);

  return state[0];
}
```
In my demo I will zip together 2 intervals of the same speed to avoid a memory leak, but stagger their start times. 
```tsx
function UseZipDemo() {
  const timer1 = useTimer(1000, 4000);
  const timer2 = useTimer(3000, 4000);
  const timer2Times10 = timer2 !== undefined ? timer2 * 10 : undefined;
  const zipped = useZip(timer1, timer2Times10);
  return (
    <div>
      <div>timer1: {timer1}</div>
      <div>timer2Times10: {timer2Times10}</div>
      <div>Zip: {JSON.stringify(zipped)?.replace(',', ', ')}</div>
    </div>
  );
}
```

# Transformation Operators
## `buffer`
```tsx
function useBuffer<T, V>(event: T, cutoffEvent: V) {
  const [{ lastBatch }, setState] = useState<{
    lastBatch: T[];
    buffer: T[];
  }>({ lastBatch: [], buffer: [] });

  useEffect(() => {
    if (event === undefined) return;
    setState((state) => ({
      ...state,
      buffer: state.buffer.concat(event),
    }));
  }, [event]);

  useEffect(() => {
    if (cutoffEvent === undefined) return;
    setState((state) => ({
      lastBatch: state.buffer,
      buffer: [],
    }));
  }, [cutoffEvent]);

  return lastBatch;
}
```

Demo:
```tsx
function UseBufferDemo() {
  const count = useInterval(700);
  const interval = useInterval(3000);
  const batch = useBuffer(count, interval);
  return (
    <div>
      <div>count: {count}</div>
      <div>interval: {interval}</div>
      <div>Batch: {JSON.stringify(batch)}</div>
    </div>
  );
}
```

## `bufferCount`
Similar to `buffer`:
```tsx
function useBufferCount<T>(event: T, size: number) {
  const [{ lastBatch }, setState] = useState<{
    lastBatch: T[];
    buffer: T[];
  }>({ lastBatch: [], buffer: [] });

  useEffect(() => {
    if (event === undefined) return;
    setState((state) => {
      const full = state.buffer.length === size;
      return {
        lastBatch: full ? state.buffer : state.lastBatch,
        buffer: full ? [] : state.buffer.concat(event),
      };
    });
  }, [event]);

  return lastBatch;
}
```
Demo:
```tsx
function UseBufferCountDemo() {
  const count = useInterval(700);
  const batch = useBufferCount(count, 5);
  return (
    <div>
      <div>count: {count}</div>
      <div>Batch: {JSON.stringify(batch)}</div>
    </div>
  );
}
```

## `bufferTime`
Similar to `buffer`:
```tsx
function useBufferTime<T>(event: T, time: number) {
  const interval = useInterval(time);
  const [{ lastBatch }, setState] = useState<{
    lastBatch: T[];
    buffer: T[];
  }>({ lastBatch: [], buffer: [] });

  useEffect(() => {
    if (event === undefined) return;
    setState((state) => ({
      ...state,
      buffer: state.buffer.concat(event),
    }));
  }, [event]);

  useEffect(() => {
    setState((state) => ({
      lastBatch: state.buffer,
      buffer: [],
    }));
  }, [interval]);

  return lastBatch;
}
```
Demo:
```tsx
function UseBufferTimeDemo() {
  const count = useInterval(700);
  const batch = useBufferTime(count, 3500);
  return (
    <div>
      <div>count: {count}</div>
      <div>Batch: {JSON.stringify(batch)}</div>
    </div>
  );
}
```

## `bufferToggle`
This operator takes in an observable and an observable factory:
```tsx
const clicks = fromEvent(document, 'click');
const openings = interval(1000);
const buffered = clicks.pipe(bufferToggle(openings, i =>
  i % 2 ? interval(500) : EMPTY
));
buffered.subscribe(x => console.log(x));
```

React hooks can't be dynamically defined like observables, but since the values emitted from the hooks are externally available, the logic of the observable factory can be defined outside as well if we use `useBuffer`. You might have to modify it with an `enabled` option. 

Write a comment if you want an example. I haven't ever seen this operator used. 

## `bufferWhen`
Similar to `bufferToggle`.

## `concatMap`
Let's convert this to hooks:
```typescript
const result$ = interval(1000).pipe(
  concatMap(i => i < 10 ? timer(2700 - i * 300) : of(null)),
);
```

This is going to be different from `concat`, because we cannot know how many inner observables will be created.

The inner observables will build up but they will all run and complete.

With `concatMap`, the outer observable might emit much more rapidly than the inner observable completes, so we need to have a queue of future inner observables as some state somewhere. Either we need to add a change to every other hook to keep track of a queue of inputs, or we need to keep this state independent and expose a function to shift queued values off when inner async tasks complete. This is what I will do. 

Here is `useQueue`:

```tsx
function useQueue<T>(value: T, enabled = true) {
  const [queue, setQueue] = useState<T[]>([]);
  const shiftQueue = useCallback(() => setQueue((q) => q.slice(1)), [setQueue]);

  useEffect(() => {
    if (value === undefined || !enabled) return;
    setQueue((s) => s.concat(value));
  }, [value]);

  return [queue, shiftQueue] as const;
}
```

And here is the demo:
```tsx
function UseConcatMapDemo() {
  const count = useInterval(700) % 30;
  const queueEnabled = count < 10;
  const [queue, shiftQueue] = useQueue(count, queueEnabled);
  useTimer(2700 - queue[0] * 300, undefined, !!queue.length, shiftQueue);
  return (
    <div>
      <div>count: {count}</div>
      <div>Queue: {JSON.stringify(queue)}</div>
    </div>
  );
}
```

If you run this you will see `queue` build up for 10 entries, then empty as each timeout completes more quickly than the last. 

After making this work, I think I need to implement an `options` parameter in each hook:
```typescript
interface Options<T> {
  enabled?: boolean;
  onComplete?: () => void;
}
```
Also, each inner observable needs to be able to output something related to the values that created it. Since the hooks are asynchronous, those values will no longer be available when it returns a value in the future. So those inputs need to be stored as state somehow to be accessed in the future. 

Maybe each hook needs to work like React Query by storing return values by the keys that were used to create the requests. Or at least every observable creation operator. 

Or maybe we should just use RxJS...