Building a Redux-like Store with RxJS `scan`

Hello! Welcome back to our advanced RxJS course.

In our previous lesson, we saw how to manage transient UI state by modeling complex form validation as a graph of interconnected streams. We used BehaviorSubject for inputs and combineLatest to derive validity states declaratively.

Today, we're scaling up from component-level state to managing the entire application's state. We will implement one of the most fundamental patterns in reactive state management: creating a Redux-like store using an action stream and the scan operator. This pattern is a powerful demonstration of how a few core RxJS concepts can build a robust, predictable state container from scratch.

1. The Redux Pattern: A Quick Refresher

As an experienced front-end developer, you're likely familiar with the Redux pattern. It provides a predictable way to manage application state based on three core principles:

1. Single Source of Truth: The entire state of your application is stored in a single object tree within a single store.
2. State is Read-Only: The only way to change the state is to emit an action, an object describing what happened.
3. Changes are made with Pure Functions: To specify how the state tree is transformed by actions, you write pure reducers.

This one-way data flow makes state changes predictable and transparent. For a quick visual recap of these ideas, the following video is excellent.

Our goal now is to implement this exact pattern using only RxJS.

2. Translating Redux Concepts to RxJS

Let's map the Redux concepts to their RxJS equivalents:

• Actions: An action is an event that occurs over time. A stream of actions can be perfectly represented by an RxJS Subject. We can push new actions onto the subject, and other parts of our system can listen to it. This subject becomes our action dispatcher.
• Reducer: A reducer is a function with the signature (currentState, action) => newState. It takes the last known state and the current action to compute the next state.
• State: The application state is not a static object but a value that changes over time in response to actions. This is a perfect fit for an Observable. Our final state$ observable will emit a new, complete state object every time an action is processed.

The key question is: how do we get from a stream of actions to a stream of states using a reducer function? The answer lies in the scan operator.

3. scan: The Heart of a Reactive Store

The scan operator is similar to the Array.prototype.reduce method, but for observables. It applies an accumulator function to each value emitted by the source observable and emits each intermediate accumulated value.

The signature is scan((accumulator, currentValue) => newAccumulator, seedValue).

When we apply this to our state management pattern:

• accumulator is the currentState.
• currentValue is the action from our action stream.
• newAccumulator is the newState returned by our reducer.
• seedValue is our initialState.

Unlike reduce, which only emits the final value when the source observable completes, scan emits the accumulated value at each step. This is crucial because we want our components to receive the new state immediately after every action, not just at the end of the application's life.

To see this difference in action, the following video provides a clear comparison.

4. Building the Store: Step-by-Step

Let's construct a simple counter store. We'll follow the structure outlined in the article "Implementing a Redux-like Pattern in Angular for State...", which provides a clear, framework-agnostic recipe.

Let's build our own version, abstracting the core logic into a reusable Store class.

Step 1: Define State and Actions

First, we define the shape of our state and the actions that can modify it using TypeScript interfaces.

// state.ts

// The shape of our application state
export interface CounterState {
  count: number;
}

// The types of actions that can be dispatched
export type CounterAction = 
  | { type: 'INCREMENT' }
  | { type: 'DECREMENT' }
  | { type: 'SET', payload: number };

Step 2: Create the Store Class

Now, we'll create a class to encapsulate our state logic. It will contain the action stream, the state stream, the initial state, and the reducer.

// store.ts
import { Subject, Observable, scan, startWith, shareReplay } from 'rxjs';
import { CounterState, CounterAction } from './state';

export class CounterStore {
  // 1. A Subject to hold the stream of actions
  private actions$ = new Subject<CounterAction>();

  // 2. The public-facing state stream
  public state$: Observable<CounterState>;

  // 3. The initial state of the application
  private initialState: CounterState = {
    count: 0,
  };

  constructor() {
    this.state$ = this.actions$.pipe(
      // 4. The `scan` operator is our reducer engine
      scan(this.reducer, this.initialState),

      // 5. Emit the initial state immediately on subscription
      startWith(this.initialState),

      // 6. Multicast the stream and replay the last value to new subscribers
      shareReplay(1)
    );
  }

  // 7. The reducer function calculates the next state
  private reducer(state: CounterState, action: CounterAction): CounterState {
    switch (action.type) {
      case 'INCREMENT':
        return { ...state, count: state.count + 1 };
      case 'DECREMENT':
        return { ...state, count: state.count - 1 };
      case 'SET':
        return { ...state, count: action.payload };
      default:
        return state;
    }
  }

  // 8. A public method to dispatch actions
  public dispatch(action: CounterAction): void {
    this.actions$.next(action);
  }
}

Key Implementation Details:

• scan(this.reducer, this.initialState): This is the core. It takes the stream of actions and, for each one, runs the reducer function, accumulating the state.
• startWith(this.initialState): When a component subscribes to state$, we want it to receive the current state immediately, not wait for the first action. startWith ensures an initial value is emitted upon subscription.
• shareReplay(1): This is a critical operator for performance and correctness.
  • It makes the source observable "hot" or "multicasted," meaning all subscribers share a single underlying subscription. Without this, each state$.subscribe() would create a new, independent scan process.
  • The (1) argument tells it to cache and replay the last emitted value (the most recent state) to any new subscriber. This solves the same problem as startWith but in a more robust way for a shared stream.

The Fireship video you watched earlier also has a great segment explaining the implementation of the reducer with scan and the importance of shareReplay.

5. Using the Store

Using the store is now straightforward. You create an instance and then either dispatch actions to it or subscribe to its state stream.

// main.ts
import { CounterStore } from './store';

const store = new CounterStore();

// Subscribe to state changes
const subscription = store.state$.subscribe(state => {
  console.log('Current count:', state.count);
});

// Dispatch actions
console.log('Dispatching INCREMENT...');
store.dispatch({ type: 'INCREMENT' });

console.log('Dispatching INCREMENT again...');
store.dispatch({ type: 'INCREMENT' });

console.log('Dispatching SET to 10...');
store.dispatch({ type: 'SET', payload: 10 });

// Expected Output:
// Current count: 0
// Dispatching INCREMENT...
// Current count: 1
// Dispatching INCREMENT again...
// Current count: 2
// Dispatching SET to 10...
// Current count: 10

This simple, powerful pattern forms the basis of many advanced state management techniques in RxJS.

Conclusion

Today, you've learned how to build a fully functional, Redux-like state management store from scratch using a few core RxJS tools. This demonstrates a key strength of RxJS: composing simple, powerful primitives to manage complex application concerns like state.

Key Takeaways:

• The Redux pattern can be implemented natively in RxJS.
• An RxJS Subject is a natural fit for an action dispatcher.
• The scan operator is the perfect engine for a reducer, as it accumulates state over time and emits each intermediate result.
• The reducer itself is a pure function with the signature (state, action) => newState.
• shareReplay(1) is essential for creating a single, shared state stream that provides the most recent state to any new subscribers, preventing redundant work and ensuring consistency.

In our next lesson, we will build upon this foundation. Now that we have a way to manage state, we'll tackle the complexities of asynchronous side effects by learning how to implement optimistic updates with a rollback strategy using concat, tap, and catchError.