rxjs - Suche in Observablen

Question

Ich versuche, ein einfaches Suchschema zu implementieren, um nach Werten in einer beobachtbaren von einer anderen beobachtbaren zu suchen. Die buildLookup Funktion baut unter einer Lookup-Tabelle Werte von einer beobachtbaren mit:

// Build lookup table from an observable. 
// Returns a promise 
function buildLookup(obs, keyName, valName) { 
    const map = new Map(); 
    obs.subscribe((obj) => map.set(obj[keyName], obj[valName])); 

    // use concat to force wait until `obs` is complete 
    return obs.concat(Observable.from([map])).toPromise(); 
} 

Dann habe ich eine andere Funktion dieses das Ergebnis dieser Funktion verwendet (ein Versprechen):

// Lookup in a previously built lookup table. 
function lookup(source, prom, keyName, fieldName) { 
    return source.map((obj) => { 
     const prom2 = prom.then((map) => { 
      return lodash.assign({}, obj, { [fieldName]: map.get(String(obj[keyName])) }); 
     }); 
     return Observable.fromPromise(prom2); 
    }) 
    .flatMap((x) => x); 
} 

Aus irgendeinem Grund, diese Implementierung funktioniert nicht, und jeder andere Lookup scheint zu scheitern. Könne Sie jemand leitet mich auf:

• was mit diesem Code ist falsch, und
• ob es ein besserer Weg, um so etwas zu implementieren?

Vielen Dank im Voraus für Ihre Hilfe!

Ich füge mein Testcode unter:

"use strict"; 
const lodash = require("lodash"); 
const rxjs = require("rxjs"); 
const chai = require("chai"); 

const Observable = rxjs.Observable; 
const assert = chai.assert; 
const assign = lodash.assign; 

describe("search",() => { 
    it("simple search", (done) => { 
     let nextId = 1, nextId2 = 1; 
     const sourceObs = Observable.interval(5).take(5).map((i) => { 
      const id = nextId++; 
      return { id: `${id}` }; 
     }); 

     const searchableObs = Observable.interval(5).take(5).map((i) => { 
      const id = nextId2++; 
      return Observable.from([ 
       { id: `${id}`, code: "square", val: id * id }, 
      ]); 
     }).flatMap((x) => x); 


     const results = []; 
     const verifyNext = (x) => { 
      assert.isDefined(x); 
      results.push(x); 
     }; 
     const verifyErr = (err) => done(err); 
     const verifyComplete =() => { 
      assert.equal(results.length, 5); 
      try { 
       results.forEach((r) => { 
        console.log(r); 
        // assert.equal(r.val, r.id*r.id); <== *** fails *** 
       }); 
      } catch (err) { 
       done(err); 
      } 
      done(); 
     }; 

     // main 
     const lookupTbl = buildLookup(searchableObs, "id", "val"); // promise that returns a map 
     lookup(sourceObs, lookupTbl, "id", "val") 
      .subscribe(verifyNext, verifyErr, verifyComplete) 
      ; 
    }); 

}); 


// output 
// { id: '1', val: 1 } 
// { id: '2', val: undefined } 
// { id: '3', val: 9 } 
// { id: '4', val: undefined } 
// { id: '5', val: 25 } 

Answer 1

Also, ein paar Dinge hier zu adressieren.

Das Hauptproblem ist, dass Sie in Ihrem sourceObs und searchableObs beobachtbaren Nebenwirkungen tun, und es wird nicht veröffentlicht, so dass die Nebenwirkungen passieren mehrmals, weil Sie mehrmals abonnieren Sie eine falsche Karte geben ganz. Zum Beispiel bekomme ich Karten wie:

{"1" => 1, "4" => 16, "7" => 49, "12" => 144}

Aber Sie tun etwas so trivial, dass Sie sollten wirklich nicht Verwendung wandelbar Variablen.

---

Um dies zu lösen, ist hier, wie Sie die richtigen Observablen erstellen:

Es

const sourceObs = Rx.Observable.range(1, 5).map(i => ({ id: `${i}` })); 

const searchableObs = Rx.Observable.range(1, 5).map(i => 
    ({ id: `${i}`, code: "square", val: i * i }) 
); 

gibt keinen Grund, eine Variable zu verwenden, da range die Zahlen zurück 1, 2, ... Und Ihre Nutzung von o.map(_ => Rx.Observable.from(...)).concatMap(e => e) ist wirklich nur das gleiche wie o ...

---

Während ich hier bin, ist dies eine vereinfachte Version Ihrer cor ist rect aber plump Funktionen:

// so buildLookup just needs to return a map once it's finished populating it 
function buildLookup(obs, keyName, valName) { 
    // following your style here, though this could be done using `scan` 
    const map = new Map(); 
    obs.subscribe((obj) => map.set(obj[keyName], obj[valName])); 
    // instead of your promise, I just wait for `obs` to complete and return `map` as an observable element 
    return obs.ignoreElements().concat(Rx.Observable.of(map)); 
} 

// and lookup just needs to wait for the map, and then populate fields in the object  
function lookup(source, prom, keyName, fieldName) { 
    return prom 
    .concatMap(map => source.map(obj => ({ obj: obj, map: map }))) 
    .map(({ obj, map }) => lodash.assign({}, obj, { [fieldName]: map.get(String(obj[keyName])) })) 
    ; 
} 

Dies sollte für Sie arbeiten.