javascript - RxJs 从 Promise 扩展数组

Question

我对 RxJ 非常陌生，需要一些帮助来了解如何实现这项工作:

  let myObs= Observable.fromPromise(this.storage.get('storageKeyForArray'))
    .map(a=>Observable.from(a))   //does this blow up if array is null?
    .map(e=>doSomethingWithElement(e)) //flatMap?
    ...
    ;

    myObs.first().subscribe((initializedArray) => {
        this.stuff= initializedArray;
        this.initialized = true;
    });

这就是我正在尝试做的事情:

1. 从本地存储加载数组
2. 对数组的每个元素执行异步查找
3. 等待所有元素查找完成
4. 合并回(现已完全初始化)数组
5. 对数组进行排序(这可以在第一步之后随时完成)

Answer 1

你需要这样的东西:

Observable.fromPromise(asyncLoadArrayOperation())
    .flatMap(x => Observable.from(x)) 
    .flatMap(x => someAsyncFunction(x))
    .toArray()
    .subscribe(x => ...)

flatMap(x => Observable.from(x)) - 假设 x 是一个数组，Observable.from 返回一个 Observable，它逐一发出数组中的项目。 flatMap 运算符将展平此 Observable，这意味着下一个运算符将获取项目而不是项目的 Observable(如果使用“map”)。

flatMap(x => someAsyncFunction(x)) - 在本例中，flatMap 用于获取 someAsyncFunction Promise 执行结果。同样，如果您使用“map”，它会发出一个 Promise。

toArray() - 该运算符收集所有项目，直到源 Observable 完成，然后发出一个项目 - 所有项目的数组。