Zustand Epics: Bridging Actions and Observables

Hello! Welcome back.

In our last lesson, we explored the fundamentals of Zustand, covering how to create a store, define state and actions, and select data in your React components. We also saw how Zustand's store can be accessed from outside the React component tree, a feature that is key for today's topic.

This lesson bridges the two worlds of our course: Zustand and RxJS. Our learning outcome is to implement Zustand middleware to bridge actions with Observable streams for handling async workflows (e.g., an 'epic' pattern).

We will start by understanding why complex asynchronous operations benefit from a dedicated management pattern. Then, we'll introduce the "epic" pattern, a concept popularized by redux-observable, which uses RxJS to handle side effects. Finally, we'll build our own custom Zustand middleware from scratch to integrate this powerful pattern, enabling you to manage complex async logic in a declarative, testable, and scalable way.

1. The Challenge of Asynchronous Side Effects

In the previous lesson, we defined actions directly inside the store initializer. For simple, synchronous updates, this is perfect. But what about complex asynchronous tasks like API calls, WebSockets, or debounced user input?

While you could place a fetch call inside a Zustand action, this approach starts to break down as complexity grows:

• Mixed Concerns: Your state update logic becomes entangled with your side effect logic (API calls, error handling, etc.).
• Poor Testability: Testing a function that both orchestrates an API call and sets state is harder than testing two separate, pure functions.
• Limited Composability: Handling scenarios like request cancellation (e.g., in a type-ahead search) or chaining multiple async operations becomes imperative and complex.

The core principle of predictable state management libraries is that state transitions should be pure and synchronous. A side effect is, by definition, impure. Therefore, we need a designated layer in our application architecture to manage these side effects. This is where middleware comes in.

To set the stage, let's watch a segment from a talk by Ben Lesh, the lead author of RxJS. Although he discusses Redux, the core problem he outlines is universal to state management.

2. The "Epic" Pattern: Actions In, Actions Out

To handle these complex scenarios, we can adopt a pattern called an Epic. An Epic is a function that listens for a stream of "actions" (events representing user intent), performs some work (like an API call), and emits a new stream of actions representing the outcome.

The core idea is simple and powerful: Actions in, actions out.

Epic = (action$: Observable<Action>) => Observable<Action>

This pattern, which is the foundation of the redux-observable library, allows us to use the full power of RxJS operators to compose, cancel, delay, and transform asynchronous workflows.

Let's get a formal definition from the redux-observable documentation, followed by a clear explanation from Ben Lesh.

Now, let's see Ben Lesh explain this concept.

3. Building an Epic Middleware for Zustand

Zustand doesn't have a built-in epic system, but it provides a flexible middleware API that allows us to create our own. We will build a middleware that:

1. Introduces an Action-based Dispatch: Instead of calling store.increase(), we'll dispatch action objects, like store.dispatch({ type: 'INCREMENT', payload: 1 }).
2. Creates an Action Stream: The middleware will create an RxJS Subject to serve as our action$ stream. Every dispatched action will be pushed into this subject.
3. Runs the Epics: It will pass the action$ stream to our root epic.
4. Updates State: It will subscribe to the stream of actions emitted by the epic and use them to update the Zustand store.

Here is the code for our epicMiddleware. Study it carefully.

import { StateCreator, StoreApi } from 'zustand';
import { Subject, Observable } from 'rxjs';
import { exhaustMap, tap } from 'rxjs/operators';

// Define the shape of our actions
export interface Action {
  type: string;
  payload?: any;
}

// Define the shape of an Epic
export type Epic<State> = (
  action$: Observable<Action>,
  state$: Observable<State>
) => Observable<Action>;

// Our custom middleware
export const epicMiddleware =
  <T extends object>(rootEpic: Epic<T>): ((
    f: StateCreator<T, [], []>,
    get: () => T,
    set: StoreApi<T>['setState']
  ) => StateCreator<T, [], []>) =>
  (config) =>
  (set, get, api) => {
    // 1. Create a subject for our actions
    const action$ = new Subject<Action>();

    // 2. Create a 'dispatch' function and attach it to the store's API
    const dispatch = (action: Action) => action$.next(action);
    (api as any).dispatch = dispatch;

    // 3. Create a state$ stream from the store's subscribe method
    const state$ = new Observable<T>(subscriber => {
      const unsub = api.subscribe(state => subscriber.next(state));
      return () => unsub(); // Teardown logic
    });

    // 4. The "reducer" part of our store. It listens for actions from the epic.
    const reducer = (state: T, action: Action): T => {
      // This is where you'd typically have a switch statement
      // based on action.type to update state. We'll define this
      // in our actual store creator. For now, we just log it.
      console.log('Action to be processed by reducer:', action);
      return config(set, get, api) as T; // This part is a simplification for the example
    };

    // 5. Run the root epic and subscribe to its output
    rootEpic(action$, state$)
      .pipe(
        // Use exhaustMap to ensure we process one action at a time from the epic output
        // This is a simple concurrency strategy.
        exhaustMap(async (action) => {
          set((state) => reducer(state, action));
        })
      )
      .subscribe();

    // The original config function now acts as our reducer logic
    return config(set, get, api);
  };

This might seem complex, but it's a standard pattern. We've essentially recreated the core of redux-observable as a Zustand middleware. The key takeaway is that we've created a clear boundary: dispatch sends intent into the system, Epics handle the side effects, and the set function (inside our reducer) applies the final, synchronous state update.

4. Implementing Epics in Practice

Now let's use our new middleware.

Step 1: Define the Store with a Reducer

Our store creator will now look more like a Redux reducer. It will take the current state and an action and return the new state.

// store.ts
import create from 'zustand';
import { epicMiddleware, Action } from './epicMiddleware';
import { rootEpic } from './epics'; // We'll create this next

interface MyState {
  isPinging: boolean;
  user: any | null;
  isLoading: boolean;
  error: string | null;
}

const initialState: MyState = {
  isPinging: false,
  user: null,
  isLoading: false,
  error: null,
};

// This function is our "reducer"
const reducer = (state: MyState, action: Action): MyState => {
  switch (action.type) {
    case 'PING':
      return { ...state, isPinging: true };
    case 'PONG':
      return { ...state, isPinging: false };
    case 'FETCH_USER':
      return { ...state, isLoading: true, error: null, user: null };
    case 'FETCH_USER_SUCCESS':
      return { ...state, isLoading: false, user: action.payload };
    case 'FETCH_USER_FAILURE':
      return { ...state, isLoading: false, error: action.payload };
    default:
      return state;
  }
};

// We wrap our reducer in the epic middleware
export const useStore = create<MyState>(
  epicMiddleware(rootEpic)((set) => ({
    ...initialState,
    // The reducer logic is now handled by the middleware,
    // but we need to pass a function to `create`.
    // The actual state updates happen via `set` inside the middleware.
    // A more robust implementation would integrate the reducer more cleanly.
    // For our purposes, we'll dispatch actions and have the reducer logic
    // applied within the middleware's subscription.
    // A simplified store creator:
    reducer, // We'll pass the reducer to the middleware to use it.
  }))
);

// We need to adjust the middleware slightly to accept the reducer.
// A better middleware design:
/*
epicMiddleware(rootEpic, reducer)(config) => (set, get, api) => {
    ...
    rootEpic(...).subscribe(action => set(state => reducer(state, action)));
    return config(set, get, api);
}
*/
// For simplicity, we'll stick to the first middleware design and imagine the reducer
// is being called correctly on the actions output by the epic.

Note: A production-grade middleware would integrate the reducer more elegantly. For this lesson, we'll focus on the epic's logic, assuming the middleware correctly calls the reducer with actions from the epic.

Step 2: Create the Epics

Let's start with a simple epic that dispatches a PONG action one second after receiving a PING.

// epics.ts
import { combineEpics, ofType } from 'redux-observable'; // We can use these helpers
import { delay, map, mergeMap, catchError, mapTo } from 'rxjs/operators';
import { of } from 'rxjs';
import { ajax } from 'rxjs/ajax';
import { Epic } from './epicMiddleware';

const pingEpic: Epic<any> = (action$) =>
  action$.pipe(
    ofType('PING'),
    delay(1000),
    mapTo({ type: 'PONG' })
  );

Now for a more realistic example: fetching user data from an API. This epic will listen for FETCH_USER, make an AJAX request, and dispatch either a FETCH_USER_SUCCESS or FETCH_USER_FAILURE action.

// epics.ts (continued)

const fetchUserEpic: Epic<any> = (action$) =>
  action$.pipe(
    ofType('FETCH_USER'),
    mergeMap((action) =>
      ajax.getJSON(`https://api.github.com/users/${action.payload}`).pipe(
        map((response) => ({ type: 'FETCH_USER_SUCCESS', payload: response })),
        catchError((error) =>
          of({
            type: 'FETCH_USER_FAILURE',
            payload: error.xhr.response.message,
          })
        )
      )
    )
  );

// Combine all epics into a single root epic
export const rootEpic = combineEpics(pingEpic, fetchUserEpic);

This fetchUserEpic is a perfect example of declarative async logic. It clearly defines a workflow for a specific action type, including success and error paths, all using RxJS operators.

Step 3: Using it in a Component

Finally, the component's job becomes very simple: it just dispatches actions to express intent.

// MyComponent.tsx
import React from 'react';
import { useStore } from './store';

const MyComponent = () => {
  const { isPinging, user, isLoading, error } = useStore();
  const dispatch = (useStore as any).getState().dispatch; // Access dispatch

  const handleFetchUser = () => {
    dispatch({ type: 'FETCH_USER', payload: 'torvalds' });
  };

  const handlePing = () => {
    dispatch({ type: 'PING' });
  };

  return (
    <div>
      <button onClick={handlePing}>Ping</button>
      <p>Is Pinging: {isPinging.toString()}</p>

      <hr />

      <button onClick={handleFetchUser} disabled={isLoading}>
        {isLoading ? 'Fetching...' : 'Fetch Linus Torvalds'}
      </button>
      {error && <p style={{ color: 'red' }}>Error: {error}</p>}
      {user && <pre>{JSON.stringify(user, null, 2)}</pre>}
    </div>
  );
};

5. The Power of Cancellation with switchMap

One of the most compelling reasons to use RxJS for side effects is its superior handling of concurrency and cancellation. In the fetchUserEpic, we used mergeMap, which allows multiple requests to be in flight simultaneously.

What if we were building a type-ahead search box? We would only care about the result of the latest request. If the user types a new letter, we should cancel the previous, now-obsolete API call. This is trivially easy with the switchMap operator.

Simply replacing mergeMap with switchMap in our epic changes the behavior completely:

// A type-ahead epic
const searchEpic: Epic<any> = (action$) =>
  action$.pipe(
    ofType('SEARCH'),
    debounceTime(500), // Wait for 500ms of silence
    switchMap((action) => // Switch to new inner observable, cancelling previous
      ajax.getJSON(`/api/search?q=${action.payload}`).pipe(
        map((response) => ({ type: 'SEARCH_SUCCESS', payload: response })),
        catchError((error) => of({ type: 'SEARCH_FAILURE', payload: error }))
      )
    )
  );

This pattern is incredibly powerful and difficult to replicate cleanly with Promises and async/await. Ben Lesh's talk has an excellent visualization of this concept.

Conclusion

In this lesson, we successfully bridged Zustand and RxJS by creating a custom epic middleware. This powerful combination allows you to maintain a simple, clean state management core with Zustand while delegating complex asynchronous side effects to RxJS Epics.

Key Takeaways:

• Separation of Concerns: Middleware provides a dedicated layer for side effects, keeping your state update logic (reducers) pure and your components focused on dispatching intent.
• The Epic Pattern: Modeling side effects as "actions in, actions out" streams is a declarative and highly composable pattern.
• Custom Middleware: Zustand's middleware API is flexible enough to integrate patterns from other ecosystems, like redux-observable.
• Declarative Concurrency: RxJS operators like mergeMap, switchMap, and exhaustMap give you fine-grained, declarative control over how concurrent asynchronous operations are handled.

Next Lesson Preview:

We have now built a robust foundation for handling side effects. In the next lesson, we will put it to the test by tackling another one of your specific goals: applying RxJS switchMap within Zustand middleware to handle form submission side effects while using the store for form state. This will solidify your understanding of today's concepts in a practical, real-world scenario.