RxJS中如何将resolve为Observable的Promise转换为目标Observable

直接对生成的高阶Observable做打平处理即可，完整实现如下：

import { from, Observable } from 'rxjs';
import { mergeAll } from 'rxjs/operators';

// 省略Item类型、apiService的上下文定义
get items$(): Observable<Item> {
    const sourcePromise: Promise<Observable<Item>> = apiService.getItemsObservable();
    const sourceObservable: Observable<Observable<Item>> = from(sourcePromise);
    return sourceObservable.pipe(mergeAll());
}

• 你当前通过from()转换得到的是一个高阶Observable：它本身不会直接发射Item类型数据，只会在Promise resolve时发射1个值——也就是API返回的Observable<Item>实例，之后立刻完成。
• mergeAll()操作符会自动订阅这个内部的Observable<Item>，把内部Observable发射的所有Item数据直接转发到下游，同时完整继承错误传递、订阅取消的逻辑，不会出现内存泄漏问题。
• 由于当前上游的高阶Observable只会发射1个内部Observable实例，你也可以根据语义偏好替换成其他打平操作符，运行效果完全一致：
  • 替换为concatAll()：按队列顺序订阅内部Observable，单内部Observable场景下和mergeAll()表现无差异
  • 替换为switchAll()：上游推送新内部Observable时自动退订旧的，单推送场景下表现一致
  • 替换为mergeMap(innerObs => innerObs)：和mergeAll()逻辑完全等价，只是写法更繁琐

不要手动在Promise的.then()回调里返回内部Observable，这种写法会破坏RxJS的事件链，无法正确处理退订、错误冒泡的逻辑。

内容的提问来源于stack exchange，提问作者Ilya Loskutov