RxJs - 2つのオブザーバブルの重複アイテムを取得

Question

2つのストリームの重複アイテムを取得する必要があります。私はほとんどそれをやっていたと思うが、第二の流れの複製であるそれらの項目が順番に行く場合にのみ、例の場合：

これは動作します：

first = Observable.of(1, 2, 3) 
second = Observable.of(2, 3, 1) 

しかし、これはしていません：

first = Observable.of(1, 4, 3) 
second = Observable.of(1, 2, 3) 

私のループが4になると、それが壊れる：

EmptyError {name: "EmptyError", stack: "EmptyError: no elements in sequence↵ at new Emp…e (http://localhost:4200/vendor.bundle.js:161:22)", message: "no elements in sequence"}

全体の私コードは1つの機能に含まれているため、コピー/貼り付けしてテストすることができます：

findDublicates() { 

    let match = 0; // setting it to 0, so later could assign other number 
    let keys = []; // list of maching keys 
    let elementAt = 0; // index of item of first observable   

    let allKeys$; 
    let validKeys$; 

    // counting the length of both observables, so this will be the number of loops 
    // that checks for dublicates 
    let allKeysLength; 
    let validKeysLength; 
    let allKeysLength$ = Observable.of(2, 1, 4, 5, 7).count() 
    allKeysLength$.subscribe(val => allKeysLength = val) 
    let validKeysLength$ = Observable.of(1, 2, 3, 8, 5).count() 
    validKeysLength$.subscribe(val => validKeysLength = val) 

    let cycles = Math.min(allKeysLength,validKeysLength); // length of the shorter observable    

    // wrapping it in a function so when called variables will take new values 
    function defineObs() { 

    allKeys$ = Observable.of(2, 1, 4, 5, 7) 
     .elementAt(elementAt).take(1); 

    validKeys$ = Observable.of(1, 2, 3, 8, 5) 
     .filter((x) => (x === match)).first(); 
    } 

    for (var i=0; i<=cycles; i++) { 

    defineObs(); 

    allKeys$.subscribe(
     function (val) { match = val }, 
     function (err) { console.log(err) }, 
     function() { console.log('Done filter')} 
    ); 
    validKeys$.subscribe(
     function (val) { keys.push(val) }, 
     function (err) { console.log(err) }, 
     function() { console.log('Done push')} 
    ); 

    elementAt += 1; 
    cycles -= 1; 

    } 

    return console.log(keys); 

} 

ありがとうございました。

Answer 1

、あなたはちょうどそれらをマージし、単一のストリーム上で重複する値を見つけることとして扱うことがあります。

first.merge(second) 
    .scan(([ dupes, uniques ], next) => 
     [ uniques.has(next) ? dupes.add(next) : dupes, uniques.add(next) ], 
     [ new Set(), new Set() ] 
    ) 
    .map(([ dupes ]) => dupes) 

注：セットscanの未定義の振る舞いを避けるために、上記は不変です。

Answer 2

私はObservable.combineLatestとscanメソッドを観察可能なシーケンスでチェックします。ここで

は私がcombineLatestを使用して2つの観測を組み合わせ、考え、その上scan演算子を適用するんですよ。一意性を保証するためにSet、さらにはmapとfilterを使用することもできます。あなたは、ストリームは、重複のセットの最初の値を放出する気にしないのであれば