我有一个 Observable,其中每个新值都应该引起一个 HTTP 请求。在客户端,我只关心最新的响应值;但是,我希望每个请求都能完成以进行监控/等。目的。
我目前拥有的是这样的:
function simulate(x) {
// Simulate an HTTP request.
return of(x).pipe(delay(6));
}
source$.pipe(
someMapFunc(x => simulate(x)),
);
当我对 someMapFunc
使用 switchMap
时,我得到了正确的响应集(只有最新的)。但是,如果请求花费的时间太长,它将被取消。
当我改用 mergeMap
时,我得到了正确的请求集(每个请求都完成了),但是我得到了错误的响应集(每个请求都完成了)。
有没有办法通过 switchMap
的响应获取 mergeMap
的请求? 我知道我可以将其编写为自定义运算符,但我想知道我是否可以使用现有/标准 rxjs 运算符构建它。总结一下我的想法:
switchMap
的一个版本,在切换时不会取消订阅;mergeMap
的一个版本,它只从最新的内部 Observable 发出值。
编辑:根据接受的答案,我得到了以下有效的答案:
function orderedMergeMap(project) {
return (s) => defer(() => {
let recent = 0;
return s.pipe(
mergeMap((data, idx) => {
recent = idx;
return project(data).pipe(filter(() => idx === recent));
})
);
});
}
最佳答案
我不是 100% 确定这是否是您所追求的,而且我还没有对此进行全面测试,但我创建了一个自定义运算符,它可能会做一些接近您所追求的事情。或许您可以稍微修改一下。
这是一个过滤掉“旧”值的mergeMap。旧值是在新源开始排放后发生的源排放。
function orderedMergeMap<T, R>(project: (v:T) => Observable<R>): OperatorFunction<T, R> {
return s => defer(() => {
let recent = 0;
return s.pipe(
map((v, i) => ({order: i, payload: v})),
mergeMap(({order, payload}) => project(payload).pipe(
map(v => ({order, payload: v}))
)),
tap(({order}) => {
if(order > recent) recent = order;
}),
filter(({order}) => order < recent),
map(({payload}) => payload)
);
});
}
OP 确定的版本:
function orderedMergeMap<T, R>(project: (v:T) => Observable<R>): OperatorFunction<T, R> {
return s => defer(() => {
let recent = 0;
return s.pipe(
mergeMap((data, idx) => {
recent = idx;
return project(data).pipe(
filter(() => idx === recent)
);
})
);
});
}
关于typescript - switchMap 结合 mergeMap,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/70073963/