Rxjs：チャンクと遅延ストリーム？要するに

Question

は、10のチャンクにチャンクに本当に大きな配列をしようとし、ここで次の10

を放出する前に5秒待っが、私は現在、

Rx.Observable 
    .from(hugeArray) 
    .bufferCount(10) 
    .delay(5000) //want to wait 5 secs 
    .flatMap(e => e) // this needs to go after to flatten the array, buffer spits out arrays of entries 
    .flatMap((data, index) => Rx.Observable.create(observer => { 
     // going to render stuff here 
     observer.onNext(data) 
     observer.onCompleted(); 

    })) 
    .subscribe(val => console.log('Buffered Values:', val)); 

ちょうど今までに10のチャンクをやろうとしなければならないものです5秒、最初の遅延を行うことができただけで、残りの部分が放出されました。

Answer 1

あなたのチェーンは一度にすべてを放出し、その後、すべてのチャンクが全く同じ瞬間に遅延が経過するように、各チャンクが同時に5秒間待つようにスケジューリングしました。

ソリューションは、Observableごとに1つずつサブスクライブするconcatMap()を使用することができます。

Rx.Observable 
    .from(hugeArray) 
    .bufferCount(10) 
    .concatMap(data => Rx.Observable.of(data).delay(5000)) 
    .flatMap(e => e) // or mergeAll() or concatAll() 
    .flatMap((data, index) => Rx.Observable.create(observer => { 
     // going to render stuff here 
     observer.onNext(data); 
     observer.onCompleted(); 
    })) 
    .subscribe(val => console.log('Buffered Values:', val));