如何使用 RxJS 运算符创建可观察的自定义对象流答案

【问题标题】：How to create an observable stream of custom objects using RxJS operators如何使用 RxJS 运算符创建可观察的自定义对象流
【发布时间】：2020-08-20 17:14:24
【问题描述】：

我目前有如下输入

const config = {
  'mainA': { sub1: { name: 'test1'}, sub2: { name: 'test2'}},
  'mainB': { sub1: { name: 'test3'}, sub2: { name: 'test4'}}
};

我正在尝试编写一个函数 (createCustomObservable)，该函数将使用标准 RsJS 运算符创建一个 observable，如下所示

var observable = createCustomObservable(config);
observable.subscribe((x) => console.log(x));

控制台输出应如下所示

{'mainA': 'test1'} -> {'mainA': 'test2'} -> {'mainB': 'test3'} ->  {'mainB': 'test4'}

一系列具有单一属性的对象

有谁知道如何使用 RxJS 操作符来实现这一点？任何帮助将不胜感激。

【问题讨论】：

标签： javascript typescript rxjs

【解决方案1】：

您要解决的主要问题是遍历对象以获取所有包含“名称”字段的对象，并获取它们的值。

没有自动执行此操作的 Rx 运算符，因此要完成此任务，您只需使用 Rx.Observable.create - https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/create.md

const config = {
'mainA': {
    sub1: {
        name: 'test1'
    },
    sub2: {
        name: 'test2'
    }
},
'mainB': {
    sub1: {
        name: 'test3'
    },
    sub2: {
        name: 'test4'
    }
}
};


function traverse(o, func) {
    for (var i in o) {
        func.apply(this, [i, o[i]]);
        if (o[i] !== null && typeof(o[i]) == "object") {
            //going on step down in the object tree!!
            traverse(o[i], func);
        }
    }
}

var source = Rx.Observable.create(function(observer) {

    traverse(config, function(key, value) {
        if (key == "name")
            observer.onNext(value);
    });

    observer.onCompleted();
    return function() {
        console.log('disposed');
    };
});

source.subscribe(function(next) {
    console.log(next);
})

示例：https://jsbin.com/yolufovubi/edit?js,console

【讨论】：

【解决方案2】：

我们可以通过 Observable 构造函数创建一个新的流，你必须手动调用 next()、error() 和 complete() 函数。

function createCustomObservable(config) {
   return new Observable(
      observer => {
         try {
           observer.next(config)
        } catch(err) {
          observer.error(err);
       } finally {
         observer.complete();
      }
    })
 }

并以这种方式使用它

var observable = createCustomObservable(config);
observable.subscribe((x) => console.log(x));

【讨论】：