Angular 5/RxJS中Observable链式调用的正确操作符及async/await实现

Hey there! Let's work through your RxJS flow issue and also cover how to implement this with async/await. Here's what you need to know:

Your core flow is: first fetch the current user, trigger the LoadUser action, then run the two list requests in parallel and trigger their respective actions. There are two common ways to handle the parallel step depending on whether you want to trigger actions as each request completes, or wait for both to finish first.

Option 1: Trigger actions after both parallel requests complete #

Use forkJoin to wait for both tagsList() and timezonesList() to emit their final value, then handle both actions at once. This is great if you need to ensure both datasets are loaded before proceeding:

import { forkJoin, tap, switchMap } from 'rxjs';

getCurrentUser().pipe(
  // Trigger LoadUser as soon as we get the user data (side effect)
  tap(user => LoadUser(user)),
  // Switch to a new observable that runs both list requests in parallel
  switchMap(() => forkJoin({
    tags: tagsList(),
    timezones: timezonesList()
  }))
).subscribe(({ tags, timezones }) => {
  LoadTags(tags);
  LoadTimezones(timezones);
});

Option 2: Trigger actions as each parallel request completes #

Use merge to subscribe to both list observables at the same time. This way, as soon as tagsList() returns, LoadTags fires immediately (no need to wait for timezones, and vice versa):

import { merge, tap, switchMap } from 'rxjs';

getCurrentUser().pipe(
  tap(user => LoadUser(user)),
  switchMap(() => merge(
    // Handle tags as soon as they're available
    tagsList().pipe(tap(tags => LoadTags(tags))),
    // Handle timezones as soon as they're available
    timezonesList().pipe(tap(timezones => LoadTimezones(timezones)))
  ))
).subscribe();

Key Operators Explained #

• tap: Used for side effects (like dispatching actions) without modifying the observable stream.
• switchMap: Switches from the getCurrentUser observable to the parallel requests observable once the user data is received.
• forkJoin: Combines the final emission of multiple observables into a single object/array.
• merge: Combines multiple observables into one, emitting values as each source observable emits.

If you prefer using async/await, you'll need to convert your observables to promises. For RxJS 7+, use firstValueFrom (the replacement for the deprecated toPromise()):

Option 1: Wait for both parallel requests to complete #

import { firstValueFrom } from 'rxjs';

async function initializeAppData() {
  // Step 1: Fetch user and trigger LoadUser
  const user = await firstValueFrom(getCurrentUser());
  LoadUser(user);

  // Step 2: Run both list requests in parallel
  const [tags, timezones] = await Promise.all([
    firstValueFrom(tagsList()),
    firstValueFrom(timezonesList())
  ]);

  // Trigger actions once both are done
  LoadTags(tags);
  LoadTimezones(timezones);
}

// Execute the function
initializeAppData();

Option 2: Trigger actions as each request completes #

If you want each action to fire as soon as its respective request finishes (without waiting for the other), wrap each request in an async IIFE inside Promise.all:

import { firstValueFrom } from 'rxjs';

async function initializeAppData() {
  const user = await firstValueFrom(getCurrentUser());
  LoadUser(user);

  // Run both requests in parallel, handling each result immediately
  await Promise.all([
    (async () => {
      const tags = await firstValueFrom(tagsList());
      LoadTags(tags);
    })(),
    (async () => {
      const timezones = await firstValueFrom(timezonesList());
      LoadTimezones(timezones);
    })()
  ]);
}

initializeAppData();

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