RxJS 可观察到函数链的结果

Question

我有一个函数getC，它依赖于getB 的输出，它依赖于getA 的输出。我想创建一个可观察的stream，这样当我订阅它时，我就能得到每个函数的输出，如sn-p所示。

RxJS 中是否有一个运算符可以提供与 sn-p 中相同的行为？还是嵌套switchMap、concat 和of 是唯一的方法？

const { concat, from, of } = rxjs
const { switchMap } = 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(
  switchMap(a => concat(
    of(a),
    from(getB(a)).pipe(
      switchMap(b => concat(
          of(b),
          getC(b)
        )
      )
    )
  ))
);

stream.subscribe(value => {
  console.log(value);
});

<script src="https://unpkg.com/rxjs@7.3.0/dist/bundles/rxjs.umd.min.js"></script>

Answer 1

如果您提前知道您的依赖项，您可以将它们硬编码到不同的流中：

1. c$ 需要来自 b$ 的最新发射
2. b$ 需要来自 a$ 的最新发射
3. 按“顺序”运行流，同时尊重它们的依赖关系，这样您最终会得到三个排放，每个排放（第一个除外）取决于前一个排放。

const a$ = of('A').pipe(delay(600));
const b$ = of('B').pipe(delay(400), combineLatestWith(a$), map(([b, a]) => a + b));
const c$ = of('C').pipe(delay(200), combineLatestWith(b$), map(([c, b]) => b + c));
concat(a$, b$, c$).subscribe(x => console.log(x));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.3.0/rxjs.umd.min.js" integrity="sha512-y3JTS47nnpKORJX8Jn1Rlm+QgRIIZHtu3hWxal0e81avPrqUH48yk+aCi+gprT0RMAcpYa0WCkapxe+bpBHD6g==" crossorigin="anonymous" referrerpolicy="no-referrer"></script>
<script>
const {of, concat} = rxjs;
const {delay, combineLatestWith, map} = rxjs.operators;
</script>

Answer 2

您可以进行一些重构以使您的函数递归

const getValue = (propertyName, value = null) => new Promise(resolve => 
  resolve(value ? {...value, [propertyName]: propertyName.toUpperCase()} : {[propertyName]: propertyName.toUpperCase()})
);

然后你可以使用神奇的 rxjs 操作符EXPAND

 import { from, EMPTY } from 'rxjs';
 import { expand } from 'rxjs/operators';


from(getValue('a')).pipe(  
 expand((previousData) => {
     if(!previousData.hasOwnProperty('b')) {
       return from(getValue('b', previousData))
     } else if(!previousData.hasOwnProperty('c')) {
       return from(getValue('c', previousData))
     } else {
       return EMPTY
     }
   }
 )
).subscribe(console.log);

这里是一个工作代码https://stackblitz.com/edit/typescript-czpotj