Question: 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;
});

```

Answer: 
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?

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; });
    });
};

```

Usage:

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

```
Not that I'm aware of. You could roll your own. I haven't tested this, but here's an idea:

```
Observable.prototype.flatFilter = function(predicate) {
  var self = this;
  return Observable.create(function(obs) {
    var disposable = new CompositeDisposable();
    disposable.add(self.subscribe(function(x) {
       disposable.add(predicate(x).subscribe(function(result) {
         if(result) {
            obs.onNext(x);
         }
       }, obs.onError.bind(obs)));
    }, obs.onError.bind(obs), obs.onCompleted.bind(obs)));
    return disposable;
  });
};

```

And you could use it like so:

```
someStream.flatFilter(function(x) {
  return Rx.DOM.get('/isValid?check=' + x);
}).subscribe(function(x) {
  console.log(x);
});

```