rxjs - 在 observables 中搜索

Question

我正在尝试实现一个简单的搜索方案，以从另一个可观察对象中搜索一个可观察对象中的值。下面的buildLookup 函数使用来自可观察对象的值构建查找表：

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

然后我有另一个函数，它使用这个函数的结果（一个承诺）：

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

由于某种原因，此实现不起作用，并且其他所有查找似乎都失败了。有人可以指导我吗：

• 这段代码有什么问题，以及
• 是否有更好的方法来实现这样的功能？

非常感谢您的帮助！

我在下面附上我的测试代码：

"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

所以，这里有很多事情要解决。

主要问题是你在你的sourceObs和searchableObs observables中做了副作用，它没有发布，所以副作用发生了多次，因为你订阅了多次，给你一个错误的地图完全。例如，我得到如下地图：

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

但是您所做的事情是如此微不足道，以至于您真的应该不使用可变变量。

---

要解决这个问题，您可以通过以下方式创建适当的 observable：

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

没有理由使用变量，因为range 返回数字 1、2、... 而你对o.map(_ => Rx.Observable.from(...)).concatMap(e => e)的使用其实和o一样...

---

当我在这里时，这是您正确但笨拙的功能的简化版本：

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

这应该适合你。