如何用RxJS方式解析传入Subject参数的Promise？

我将Promise作为参数传入RxJS的Subject：

const work = new Subject<{ id: number; dialogRef: Promise<typeof Dialog> }>();

后续想要使用该Promise内部的实例：

...
exhaustMap(({ id, dialogRef }) =>
  http.get({ id }).pipe(tap(() => dialogRef.close()))
)
...

但dialogRef是Promise，必须先解析才能调用close()方法。我目前可以用async/await实现，但希望采用RxJS原生方式解决，想找到能解析Promise并将实例传递给exhaustMap的方法，比如类似这样的操作：

resolve(({ dialogRef }) => dialogRef),
exhaustMap(({ id, dialogRef }) => //<-- dialogRef 不再是Promise，而是实例
      http.get({ id }).pipe(tap(() => dialogRef.close()))
 )

能否用RxJS实现该需求？

示例代码：

import { of, Subject } from 'rxjs';
import { exhaustMap, tap } from 'rxjs/operators';
console.clear();

const http = {
  get: ({ id }) => of(`data: ${id}`),
};

const Dialog = {
  close: () => {
    console.log('in close!');
  },
};

const work = new Subject<{ id: number; dialogRef: Promise<typeof Dialog> }>();

work
  .pipe(
    exhaustMap(({ id, dialogRef }) =>
      http.get({ id }).pipe(tap(async () => (await dialogRef).close()))
    )
  )
  .subscribe((r) => {
    console.log({ r });
  });

work.next({ id: 1, dialogRef: Promise.resolve(Dialog) });

当然可以用RxJS原生操作符实现，核心是利用from()将Promise转换为Observable，再通过流操作传递解析后的实例。

方案1：提前解析Promise（在exhaustMap之前） #

使用switchMap配合from()解析Promise，同时保留原有的id参数：

import { of, Subject, from } from 'rxjs';
import { exhaustMap, tap, switchMap, map } from 'rxjs/operators';
console.clear();

const http = {
  get: ({ id }) => of(`data: ${id}`),
};

const Dialog = {
  close: () => {
    console.log('in close!');
  },
};

const work = new Subject<{ id: number; dialogRef: Promise<typeof Dialog> }>();

work
  .pipe(
    switchMap(({ id, dialogRef }) => 
      from(dialogRef).pipe(
        map(dialogInstance => ({ id, dialogInstance }))
      )
    ),
    exhaustMap(({ id, dialogInstance }) =>
      http.get({ id }).pipe(tap(() => dialogInstance.close()))
    )
  )
  .subscribe((r) => {
    console.log({ r });
  });

work.next({ id: 1, dialogRef: Promise.resolve(Dialog) });

方案2：在exhaustMap内部解析Promise #

如果不想额外增加操作符层级，也可以在exhaustMap里直接处理Promise：

import { of, Subject, from } from 'rxjs';
import { exhaustMap, tap, switchMap } from 'rxjs/operators';
console.clear();

const http = {
  get: ({ id }) => of(`data: ${id}`),
};

const Dialog = {
  close: () => {
    console.log('in close!');
  },
};

const work = new Subject<{ id: number; dialogRef: Promise<typeof Dialog> }>();

work
  .pipe(
    exhaustMap(({ id, dialogRef }) =>
      from(dialogRef).pipe(
        switchMap(dialogInstance => 
          http.get({ id }).pipe(tap(() => dialogInstance.close()))
        )
      )
    )
  )
  .subscribe((r) => {
    console.log({ r });
  });

work.next({ id: 1, dialogRef: Promise.resolve(Dialog) });

关键说明 #

• from()是RxJS内置工具，可将Promise、数组、迭代器等转换为Observable，Promise resolved时会发出对应的值，rejected时会触发错误流。
• 两种方案都完全基于RxJS的流操作处理异步逻辑，避免了async/await的使用，更贴合RxJS的编程范式。

内容的提问来源于stack exchange，提问作者Jon Sud