是否存在“异步” RxJs中过滤运算符的版本？ [英] Is there an "async" version of filter operator in RxJs?

问题描述

我需要通过检查某个Web服务的条目来过滤observable发出的条目。正常的observable.filter运算符在这里不合适，因为它期望谓词函数同步返回判定，但在这种情况下，判决只能异步检索。

I need to filter entries emitted by an observable by checking the entry against some web service. The normal observable.filter operator is not suitable here, as it expects the predicate function to return the verdict synchronously, but in this situation, the verdict can only be retrieved asynchronously.

我可以通过以下代码进行转换，但我想知道是否有一些更好的运算符我可以用于这种情况。

I can make shift by the following code, but I was wondering whether there is some better operator I can use for this case.

someObservable.flatmap(function(entry) {
  return Rx.Observable.fromNodeCallback(someAsynCheckFunc)(entry).map(function(verdict) {
    return {
      verdict: verdict,
      entry: entry
    };
  });
}).filter(function(obj) {
  return obj.verdict === true;
}).map(function(obj) {
  return obj.entry;
});

推荐答案

以下是使用现有方法实现此类运算符的方法运营商。你需要考虑一个障碍。由于您的过滤操作是异步的，因此新项目的到达速度可能比过滤操作可以处理的速度快。在这种情况下会发生什么？是否要按顺序运行过滤器并确保维护项目的顺序？您想要并行运行过滤器并接受您的商品可能以不同的顺序出现吗？

Here's how you'd implement such an operator using existing operators. There is one snag you need to think about. Because your filter operation is async, it is possible for new items to arrive faster than your filter operation can process them. What should happen in this case? Do you want to run the filters sequentially and guarantee that the order of your items is maintained? Do you want to run the filters in parallel and accept that your items may come out in different order?

以下是运营商的2个版本

Here are the 2 versions of the operator

// runs the filters in parallel (order not guaranteed)
// predicate should return an Observable
Rx.Observable.prototype.flatFilter = function (predicate) {
    return this.flatMap(function (value, index) {
        return predicate(value, index)
            .filter(Boolean) // filter falsy values
            .map(function () { return value; });
    });
};

// runs the filters sequentially (order preserved)
// predicate should return an Observable
Rx.Observable.prototype.concatFilter = function (predicate) {
    return this.concatMap(function (value, index) {
        return predicate(value, index)
            .filter(Boolean) // filter falsy values
            .map(function () { return value; });
    });
};

用法：

var predicate = Rx.Observable.fromNodeCallback(someAsynCheckFunc);
someObservable.concatFilter(predicate).subscribe(...);

这篇关于是否存在“异步” RxJs中过滤运算符的版本？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！