18

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;
});
2
  • Can you get thte code example to a (non)working state, something copy/paste-able we can work from to help you? Commented Feb 13, 2015 at 10:40
  • I don't think it exists as an operator in the RxJs distribution. However, you can easily take the above code and turn it into your own reusable filterAsync operator.
    – Brandon
    Commented Feb 13, 2015 at 16:10

5 Answers 5

17

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(...);
3
  • 1
    Bens's solution works, but I think Brandon's is more clear. So I choose this.
    – victorx
    Commented Feb 18, 2015 at 8:31
  • 1
    Sweet, thank you. I guess this is still the most up-to-date solution?
    – Pipo
    Commented Sep 17, 2015 at 17:22
  • I added a more up-to-date version for RxJS 6 below :) Commented Aug 22, 2018 at 12:12
11

Update for RxJS 6+

Since RxJS version 6.0 we have pipe operators instead of observable prototype method chaining.

So I updated the original code of this request to RxJS 6 pipeline style improved by the information in the accepted answer.


Update 2

I released version 1.0.0 of this package:
https://www.npmjs.com/package/filter-async-rxjs-pipe/v/1.0.0
The package seems to have between 90 and 300 downloads a week now, so it seems pretty stable and I'm confident, that it will do it's job.


Update 1

I now refactored this code into a npm package.
https://www.npmjs.com/package/filter-async-rxjs-pipe

The serial variant with concatMap already works correctly, the parallel variant with flatMap seems not to run in parallel currently. But since I need the concatMap version, I currently have all I need. If somebody has an idea on how to write the parallel version correctly, please add an issue at the connected Git repository. :)


Note
Since I only need to pass a predicate function which returns a Promise, I wrote the conversion of the Promise to an Observable directly into the filterAsync method. If you need to have an Observable as filter input, feel free to adjust the code.

export function filterAsync<T>(predicate: (value: T, index: number) => Promise<boolean>): MonoTypeOperatorFunction<T> {
    let count = 0;
    return pipe(
        // Convert the predicate Promise<boolean> to an observable (which resolves the promise,
        // Then combine the boolean result of the promise with the input data to a container object
        concatMap((data: T) => {
            return from(predicate(data, count++))
                .pipe(map((isValid) => ({filterResult: isValid, entry: data})));
        }),
        // Filter the container object synchronously for the value in each data container object
        filter(data => data.filterResult === true),
        // remove the data container object from the observable chain
        map(data => data.entry)
    );
}

Here is a gist with the full ts file code, including imports:
https://gist.github.com/bjesuiter/288326f9822e0bc82389976f8da66dd8#file-filter-async-ts

1
  • Nice answer. Can't imagine why RxJs don't implement it out of the box. Commented Apr 25, 2019 at 14:01
4

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);
});
4
  • lol... I have no idea why I called it "flatFilter", I liked having it with "flatMap" I guess. ;)
    – Ben Lesh
    Commented Feb 13, 2015 at 18:13
  • Ben, thanks for the solution. However, it seems to me the first obs.onCompleted.bind(obs) is not quite right. It seems the completion of the observable returned by predicate(x) will incorrectly complete the observable returned by flatFilter(). Am I wrong?
    – victorx
    Commented Feb 16, 2015 at 9:43
  • Yeah, that was an oversight. The first onCompleted can be removed. I'd leave the onError in there though.
    – Ben Lesh
    Commented Feb 16, 2015 at 17:51
  • 1
    You've unintentionally given it switch semantics with your inner disposable. This is probably easiest implemented with flatMap or concatMap rather than rolling it by hand...
    – Brandon
    Commented Feb 17, 2015 at 1:06
0

Here's a pipe-able operator solution I think is pretty clean.

function flatFilter<T>(
    predicate: (value: T, index: number) => Observable<boolean>,
    thisArg?: unknown): MonoTypeOperatorFunction<T> {
  return flatMap(
      (v, i) => predicate.call(thisArg, v, i).pipe(filter(b => b), mapTo(v)));
}

concatMap or switchMap can be easily swapped in (flatMap is mergeMap).

0

My simple realization


export const asyncFilter = <V>(callback: (v: V) => Promise<boolean>) => {
    return (o$: Observable<V>): Observable<V> => {
        return o$.pipe(
            concatMap(v => from(callback(v)).pipe(map(r => [v, r] as const))),
            filter(([, r]) => !!r),
            map(([v]) => v)
        );
    }
}

And test for it

import { of } from "rxjs";
import { asyncFilter } from "./asyncFilter";
import { stub } from "sinon";
import { expect } from "chai";

describe('asyncFilter', () => {

    // Simple promise func
    const can = (v: number) => Promise.resolve(v % 2 === 0);


    it('should filter stream by async function', (done) => {
        const nextStub = stub();

        of(1,2,3,4,5,6).pipe(asyncFilter(v => can(v)))
            .subscribe({
                next: nextStub,
                complete: () => {
                    expect(nextStub.args.map(args => args[0])).to.eql([2,4,6])
                    done();
                }
            })
    })
})

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.