如何用RxJS正确实现用户资源Observable的动态数组更新？

Hey there! I totally get where you're coming from—when you're trying to maintain a list of users that updates with newUser or updateUser events, it's frustrating when only the first event sticks. The root problem here is that you're not persisting the state of your user array between events. Luckily, RxJS has the perfect operator for this job: scan.

Why Your Current Setup Isn't Working #

Right now, you're probably mapping each event to a new array, but there's no mechanism to carry over the previous array state. So each new event just generates a fresh array (instead of updating the existing one), which is why only the first newUser event seems to take effect.

The Solution: Use scan to Maintain State #

The scan operator works like a "reducer" (if you're familiar with Redux) — it takes the current accumulated state, applies the latest event to it, and emits the new state. This way, every subsequent event builds on the last version of your user array.

Here's a clean, scalable implementation:

import { merge } from 'rxjs';
import { map, scan } from 'rxjs/operators';

// Assume these are your existing event observables
const newUser$ = /* Observable that emits new user objects (e.g., { id: 1, name: "Alice" }) */;
const updateUser$ = /* Observable that emits update objects (e.g., { userId: 1, data: { name: "Alice Smith" } }) */;

// Combine events into a single stream with action types for clarity
const userActions$ = merge(
  newUser$.pipe(map(user => ({ type: 'NEW_USER', payload: user }))),
  updateUser$.pipe(map(update => ({ type: 'UPDATE_USER', payload: update })))
);

// Use scan to maintain and update the user array state
const users$ = userActions$.pipe(
  scan((currentUsers, action) => {
    switch (action.type) {
      case 'NEW_USER':
        // Add the new user to the array (immutably!)
        return [...currentUsers, action.payload];
      case 'UPDATE_USER':
        // Find the user to update and return a new array with the modified user
        return currentUsers.map(user => 
          user.id === action.payload.userId 
            ? { ...user, ...action.payload.data } 
            : user
        );
      default:
        // Fallback: return the current state if the action type is unknown
        return currentUsers;
    }
  }, []) // Initial state: empty user array
);

Key Details to Note #

• Immutability: We're creating new arrays/objects instead of mutating the existing ones ([...currentUsers] and { ...user }). This is crucial for RxJS to detect state changes and emit updates correctly.
• Action Types: Using a type field to distinguish between newUser and updateUser events makes the code easy to extend later (if you add deleteUser events, for example).
• Initial State: The second argument to scan is the starting state—here, we start with an empty array.

Simplified Alternative (If You Prefer Less Boilerplate) #

If you don't want to use action types, you can handle the events directly in the merge:

import { merge } from 'rxjs';
import { map, scan } from 'rxjs/operators';

const users$ = merge(
  // Pass new users through as-is
  newUser$,
  // Mark update events to distinguish them
  updateUser$.pipe(map(update => ({ isUpdate: true, ...update })))
).pipe(
  scan((currentUsers, event) => {
    if ('isUpdate' in event) {
      // Handle update
      return currentUsers.map(user => 
        user.id === event.userId ? { ...user, ...event.data } : user
      );
    } else {
      // Handle new user
      return [...currentUsers, event];
    }
  }, [])
);

Either way, the core idea is the same: scan keeps track of your user array state over time, so every newUser or updateUser event modifies the latest version of the array instead of starting from scratch.

内容的提问来源于stack exchange，提问作者OKey