如何像使用 Promises 或 async/await 一样链接 observables？

Question

假设我使用 async/await 的代码运行良好：

async function getMeSomething() : Promise<Something> {
  // some async code
}

async function getMeStuff() : Promise<Stuff> {
  // some async code
}

async function getMeSomethingWithStuff: Promise<SomethingWithStuff> {
  const something: Something = await getMeSomething();
  const stuff: Stuff = await getMeStuff();
  return new SomethingWithStuff(something, stuff);
}

这段代码可以有效地用 Promise 编写，就像任何使用 async/await 的代码一样（只是更长）：

// this is strictly equivalent to previous example, including exception management
function getMeSomethingWithStuff: Promise<SomethingWithStuff> {
  let something;
  return getMeSomething().then((res) => {
    something = res;
    return getMeStuff();
  }).then((stuff) => {
    return new SomethingWithStuff(something, stuff);
  });
}

当调用第三个函数时，它会调用第一个和第二个，合并结果并以第三个函数的调用者不知道调用了哪些底层函数的方式返回最终结果：

const somethingWithStuff: SomethingWithStuff = getMeSomethingWithStuff();
// we have the result, we don't care how we got it, the logic is completely encapsulated

我们如何仅使用 rxjs 编写等效代码？

// the following two functions are already defined and we can't redefine them
function getMeSomething() : Observable<Something> {
  // some async code
}

function getMeStuff() : Observable<Stuff> {
  // some async code
}

// we want to define this function that should return an observable
function getMeSomethingWithStuff: Observable<SomethingWithStuff> {
  // what's the code ?
}

规则：

• 不得使用 Promise、async/await 或可观察对象与 Promise 之间的转换作弊，只能使用 rxjs
• 必须正确处理异常

这似乎是一个微不足道的问题，但尽管仔细阅读了 rxjs 的文档和大量教程，但我自己却找不到答案。

Answer 1

您需要使用像 forkJoin 这样的函数来并行触发多个 observable，并使用像 switchMap 这样的高阶映射运算符之一来从一个 observable 映射到另一个。

有关forkJoin 函数和switchMap 运算符的简要说明，请参见here。

选项 1：并行

// the following two functions are already defined and we can't redefine them
function getMeSomething() : Observable<Something> {
  // some async code
}

function getMeStuff() : Observable<Stuff> {
  // some async code
}

function getMeSomethingWithStuff: Observable<SomethingWithStuff> {
  return forkJoin({
    something: getMeSomething(),
    stuff: getMeStuff()
  }).pipe(
    switchMap((res: any) => 
      new SomethingWithStuff(res.something, res.stuff)
    )
  );
}

选项 2：串行

使用多个高阶映射运算符

function getMeSomethingWithStuff: Observable<SomethingWithStuff> {
  return getMeSomething().pipe(
    switchMap(something => getMeStuff().pipe(
      map(stuff => ({ something, stuff })
    ),
    switchMap((res: any) => 
      new SomethingWithStuff(res.something, res.stuff)
    )
  );
}

更新：添加串行可观察触发器