如何在RxJS中消除订阅与错误回调重复代码以封装加载状态

使用场景 #

我有一个由RxJS Observable启动的加载操作，希望在操作结束时结束/封装该加载状态。Observable要么返回错误，要么返回新数据。
结束既指Observable返回错误，也指它触发新值。
补充说明：在下面的变体中，我还有另一个Observable来触发整个加载流程。

最小可复现示例 #

可以通过复制代码简单实现，但存在代码重复：

const {
  from,
  concat,
  throwError
} = rxjs;


concat(
  from([1, 2, 3]),
  throwError("fail")
).subscribe((x) => {
    console.log("loading end", x);
  },
  (x) => {
    console.log("loading end", x);
  }
)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.5.7/rxjs.umd.min.js"></script>

我本质上是想找一个操作符来消除上述示例中的代码重复。

显示加载状态 #

显示加载状态自然要在Observable执行前，比如用tap操作符：

const {
  from,
  concat,
  throwError,
  pipe,
  tap
} = rxjs;


concat(
  from([1, 2, 3]),
  throwError("fail")
).pipe(
   tap(() => console.log("show loading"))
).subscribe((x) => {
    console.log("loading end", x);
  },
  (x) => {
    console.log("loading end", x);
  }
)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.5.7/rxjs.umd.min.js"></script>

不过tap触发的显示加载操作不会在错误时执行。

实际场景示例 #

但在我的实际实现中不会出现这种情况，因为我合并了两个Observable，且有一个触发Observable来启动整个流程：

const {
  from,
  concat,
  throwError,
  pipe,
  tap,
  switchMap,
  Subject
} = rxjs;

// this is the trigger to load stuff (in Angular triggered from the outside component e.g.)
loadData$ = new Subject();

loadData$ //< <- this is irrelevant now and just the trigger
  .pipe(
    tap(() => console.log("show loading")), // <-- show loading when/before data loading starts
    switchMap(() => {
      return concat(
        from([4, 5, 6]),
        throwError("fail")
      )
    })
  ).subscribe((x) => { // <-- end loading when/after data loading ends or fails
      console.log("loading end", x);
    },
    (x) => {
      console.log("loading end", x);
    }
  )


loadData$.next(); // just trigger for testing here

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.5.7/rxjs.umd.min.js"></script>

尝试过的方法 #

• 我试过finalize操作符，但它只在完成时触发（无论是否出错），因此不适用。
• 我认为complete回调也会有同样的效果。

发现的问题 #

在撰写本文时，我意识到如果Observable出错，后续无法发射新值。（实际情况是否如此？）

const {
  from,
  concat,
  throwError,
  of
} = rxjs;

concat(
  from([1, 2, 3]),
  throwError("fail"),
  of ("another value") // <-- is not triggered
).subscribe((x) => {
    console.log("loading end", x);
  },
  (x) => {
    console.log("loading end", x);
  }
)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.5.7/rxjs.umd.min.js"></script>

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