如何在 rxjs 中执行异步/等待逻辑？答案

【问题标题】：How to do async/await logic in rxjs?如何在 rxjs 中执行异步/等待逻辑？
【发布时间】：2022-11-20 03:57:14
【问题描述】：

我想做这个简单的逻辑：

我有返回值的可观察对象
我想使用该值并依次运行另外两个可观察对象

我想在两个连续完成后基于我的第一个可观察值返回值

这是我尝试用 cmets 解决我的问题

updateAvatar(
    @Headers() headers: { authorization: string },
    @CurrentUserId() currentUserId: string,
    @UploadedFile() avatarFile: Express.Multer.File,
  ): Observable<{ avatarUrl: string }> {
    const obs1 = this.queue.send(
      PostCommands.uploadImage,
      new UploadImageCommandRequst(
        currentUserId,
        avatarFile.originalname,
        'avatars',
        true,
      ),
    ); // returns observable
    const obs2 = obs1.pipe(
      map(({ imageUploadUrl, imageUrl }) => {
        // need await here
        this.httpService.put(imageUploadUrl, avatarFile.buffer); // returns observable
        // also need await here. patch must be executed after put
        this.httpService.patch(
          `${this.config.GATEWAY_URL}/user/profile`,
          {
            avatarUrl: imageUrl,
          },
          { headers: { authorization: headers.authorization } },
        ); // returns observable
        // value must be returned after patch executed
        return {
          avatarUrl: imageUrl,
        };
      }),
    );
    return obs2;
  }

【问题讨论】：

你看过switchMap和mergeMap运营商了吗？
@stealththeninja 是的，我有。但我不明白如何在我的案例中使用它们

标签： rxjs

【解决方案1】：

我认为需要 switchMap 或 mergeMap。重要的区别是 switchMap 将在 obs1 发出更新的事件时重新启动，并且只发出最新事件。如果您想通过 obs1 从所有事件中发出，那么 mergeMap 可能就是您想要的。

switchMap 操作文档：https://rxjs.dev/api/index/function/switchMap
合并地图运算符文档：https://rxjs.dev/api/index/function/mergeMap

您还可以玩这个：它嵌套以访问imageUploadUrl 和imageUrl 参数，您可以将它们从可观察对象 (A) 中映射出来并稍微压平管道。

const obs2 = obs1.pipe(
  switchMap(({ imageUploadUrl, imageUrl }) =>
    //
    // (A) Projecting event from obs1 to a new observable:
    // PUT request to image upload URL.
    //
    this.httpService.put(imageUploadUrl, avatarFile.buffer).pipe(
      // 
      // (B) After observable (A) emits, we PATCH user profile
      // 
      switchMap(() => this.httpService.patch(
          `${this.config.GATEWAY_URL}/user/profile`,
          { avatarUrl: imageUrl, },
          { headers: { authorization: headers.authorization } },
        )
      ),
      //
      // (C) Finally, we wanted to return the avatar URL. We map it
      // so this is what is resolved by (A)
      //
      map(() => ({ avatarUrl: imageUrl, })),
    )
  ),
);

【讨论】：