RxJs 6:管理数组迭代和延迟

Question

我需要帮助管理数组迭代的延迟。

关于我https://jsfiddle.net/mlefree/vrL813j2/93/，两个问题：

1. 如何在每个迭代操作上添加延迟
2. 如何减少所有迭代计算值

``

...
const arrayAsObservable = of(null).pipe(
delay(500),
switchMap(_ => getObjectWithArrayInPromise()),
map(val => {
log('array', val);
return (val.myArray);
}),
switchMap(val => from(val))
);
const eachElementAsObservable = arrayAsObservable.pipe(
delay(500), // Not working : we want to wait 500ms more for each value
map(val => {
log('value', val);
return val ;
}),
switchMap(val => getNewValueInPromise(val)),
map(val => {
// Not working : why not all new values ?
log('value after computing (KO)', val);
return (val);
})
);
const summarizeAsObservable = eachElementAsObservable.pipe(
// Not working : we want to sum all new values
map(val => { 
log('value before reduce (KO)', val);
return val ;
}),
reduce((a,b) => a + b)
);
summarizeAsObservable.subscribe(msg => {
log('We did a total of (KO)', msg);
});

```

Answer 1

总的来说，代码有点太复杂了，有几行代码成为了代码的根本问题。

计算(KO(后只收到一个值的原因是您使用了switchMap，一旦源发出，它将取消订阅内部可观测值，因此您总是得到最后一个发出的值。我还将延迟更改为timer和mapTo发射值

const eachElementAsObservable = arrayAsObservable.pipe(
concatMap(value => timer(1500).pipe(mapTo(value))), // Not working : we want to wait 500ms for each value
map(val => {
log('value', val);
return val;
}),
mergeMap(val => from(getNewValueInPromise(val))),
map(val => {
// Not working : why not all new values ?
console.log('value after computing (KO)', val);
return (val);
})
);

这是一把更新过的小提琴https://jsfiddle.net/fancheung/vrL813j2/109/代码没有完全工作，因为你在承诺中通过拒绝抛出了一个错误，这将导致可观察到的停止发射，你需要在流中的某个地方放一个catchError