RxJS 的 flatMapLatest 将最新的(只有一个)嵌套的 Observable 扁平化。我有一个用例,我不想要 flatMap(将过去所有嵌套的 Observable 压平),我不想要 flatMapWithConcurrency(因为它有利于旧的 Observable,而不是最新的 Observable),所以我想要的是 flatMapLatestTwo 或flatMapLatest 的某些版本,您可以在其中指定并发嵌套 Observable 的最大数量,例如flatMapLatest(2, selectorFn)
。
这是我想要的输出(_X
指的是嵌套的 Observable X
而 eX
指的是它的第 X 个 onNext 事件):
_0e0
_0e1
_1e0
_0e2
_1e1
_2e0
_1e2
_2e1
_3e0
_2e2
_3e1
_4e0
_3e2
_4e1
_3e3
_4e2
_4e3
这是 flatMapLatest 产生的:
_0e0
_0e1
_1e0
_1e1
_2e0
_2e1
_3e0
_3e1
_4e0
_4e1
_4e2
_4e3
我更喜欢使用现有运算符的解决方案,而不是实现这个低级别的解决方案。
最佳答案
这看起来很幼稚。我正在寻找改进的方法,但它是:
Rx.Observable.prototype.flatMapLatestN = function (count, transform) {
let queue = [];
return this.flatMap(x => {
return Rx.Observable.create(observer => {
let disposable;
if (queue.length < count) {
disposable = transform(x).subscribe(observer);
queue.push(observer);
}
else {
let earliestObserver = queue[0];
if (earliestObserver) {
earliestObserver.onCompleted();
}
disposable = transform(x).subscribe(observer);
queue.push(observer);
}
return () => {
disposable.dispose();
let i = queue.indexOf(observer);
queue.splice(i, 1);
};
});
});
};
测试:
function space(n) {
return Array(n+1).join(' ');
}
Rx.Observable
.interval(1000)
.take(6)
.flatMapLatestN(2, (x) => {
return Rx.Observable
.interval(300)
.take(10)
.map(n => `${space(x*4)}${x}-${n}`);
})
.subscribe(console.log.bind(console));
它会输出:
0-1
0-2
0-3
1-0
0-4
1-1
0-5
1-2
1-3
2-0
1-4
2-1
1-5
2-2
2-3
3-0
2-4
3-1
2-5
3-2
3-3
4-0
3-4
4-1
3-5
4-2
4-3
5-0
4-4
5-1
4-5
5-2
4-6
5-3
4-7
5-4
4-8
5-5
4-9
5-6
5-7
5-8
5-9
关于javascript - RxJS如何实现flatMapLatest 二,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/32233408/