Esiste una versione "asincrona" dell'operatore filtro in RxJs?

Question

Ho bisogno di filtrare le voci emesse da un osservabile controllando la voce su alcuni servizi web. Il normale operatore observable.filter non è adatto qui, poiché si aspetta che la funzione predicato restituisca il verdetto in modo sincrono, ma in questa situazione il verdetto può essere recuperato solo in modo asincrono.

Posso eseguire il passaggio con il seguente codice, ma mi chiedevo se ci sia un operatore migliore che posso usare per questo caso.

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 1

Ecco come si sarebbe implementare un tale operatore che utilizza operatori esistenti. C'è un ostacolo che devi pensare. Poiché l'operazione di filtro è asincrona, è possibile che i nuovi elementi arrivino più velocemente di quanto l'operazione di filtro possa elaborarli. Cosa dovrebbe accadere in questo caso? Vuoi eseguire i filtri in sequenza e garantire che l'ordine dei tuoi articoli sia mantenuto? Vuoi eseguire i filtri in parallelo e accetti che i tuoi articoli possano uscire in ordine diverso?

Ecco le 2 versioni del gestore

// 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(...); 

Answer 2

Non che io sia a conoscenza. Potresti arrotolare il tuo. Non ho ancora testato questo, ma ecco un'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; 
    }); 
}; 

e si potrebbe usare in questo modo:

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