RxJS如何实现依赖函数链输出每步结果，有没有替代嵌套switchMap的方案？

结论 #

完全不需要嵌套switchMap+concat+of的写法，RxJS提供了expand递归操作符可以完美匹配你的需求，代码更简洁，可扩展性也更强。

实现方案 #

你需要的效果是按依赖顺序执行异步函数，同时依次发射每一步的返回结果，expand操作符的特性刚好匹配：它会先把当前值发射到下游，再把当前值作为入参调用回调生成下一个Observable，递归执行直到返回空Observable。

示例代码

const { from, of } = rxjs
const { expand, take } = rxjs.operators

// 原有异步函数保持不变
const getA = async() => new Promise(resolve => setTimeout(() => resolve({
  a: 'A'
}), 500));

const getB = async value => new Promise(resolve => setTimeout(() => resolve({
  a: value.a,
  b: 'B',
}), 500));

const getC = async value => new Promise(resolve => setTimeout(() => resolve({
  a: value.a,
  b: value.b,
  c: 'C',
}), 500));

const stream = from(getA()).pipe(
  // 递归处理每一步
  expand((prevResult, stepIndex) => {
    if (stepIndex === 0) return from(getB(prevResult))
    if (stepIndex === 1) return from(getC(prevResult))
    // 所有步骤执行完后返回空Observable终止递归
    return of()
  }),
  // 总共发射3个结果就结束流
  take(3)
)

stream.subscribe(console.log)

运行结果和你提供的嵌套写法完全一致，每隔500ms依次输出{a: 'A'}、{a: 'A', b: 'B'}、{a: 'A', b: 'B', c: 'C'}。

扩展优化

如果后续有更多链式依赖的异步函数，可以直接把函数整理成数组，动态生成流，不需要修改管道逻辑：

// 按依赖顺序排列异步函数，第一个函数不需要入参
const asyncTasks = [getA, getB, getC]

const stream = from(asyncTasks[0]()).pipe(
  expand((prev, idx) => idx < asyncTasks.length - 1 ? from(asyncTasks[idx + 1](prev)) : of()),
  take(asyncTasks.length)
)

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