RxJs'de "async" filtre operatörünün sürümü var mı?

Question

Bazı web hizmetlerine karşı girişi denetleyerek, gözlemlenebilir tarafından yayılan girdileri filtrelemem gerekiyor. Normal gözlemlenebilir.filter operatörü, yükleme işlevinin eşzamanlı olarak karar vermesini beklediği için burada uygun değildir, ancak bu durumda, karar yalnızca eşzamansız olarak alınabilir.

Aşağıdaki kodla geçiş yapabilirim, ancak bu dava için kullanabileceğim daha iyi bir operatör olup olmadığını merak ediyordum.

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

Burada mevcut operatörleri kullanarak bu tür bir operatör uygulamak istiyorum nasıl. Düşünmen gereken tek bir sapık var. Filtre işleminiz uyumsuz olduğundan, yeni öğelerin filtre işleminizden daha hızlı gelmesi mümkündür. Bu durumda ne olmalı? Filtreleri sırayla çalıştırmak ve öğelerinizin sırasının korunmasını garanti etmek ister misiniz? Filtreleri paralel olarak çalıştırmak ve öğelerinizin farklı bir sırada çıkabileceğini kabul etmek ister misiniz? İşte

operatörün 2 sürümleridir

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

Kullanımı:

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

Answer 2

Bildiğim kadarıyla değil. Kendi başını alabilirsin. Bu test, ama işte bir fikir değil:

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

Ve bunu gibi kullanabilirsiniz:

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