在 RxJs 计时器中覆盖发射 Observable

Question

前言

我目前正在开发的 Angular 应用程序每 10 秒从底层 API 请求一个特定状态。根据用户的状态，应该显示一系列覆盖，指示他们需要执行的步骤以使系统顺利运行。我要描述的问题主要是一个rxjs相关的问题。不过让我顺口介绍一下。请注意，整个系统要复杂得多，我正在尽最大努力提供所有必要的信息，而不会实际披露我的实际代码的一部分。

更新

我有一个使用 rx 计时器实现的每 10 秒发送一次请求的服务。计时器本身使用以下功能启动。每个周期都会向服务器发送一个 HTTP 请求，请求当前系统状态。结果被传递给一个公开可用的 BehaviorSubject，以便其他组件可以订阅它以接收值并根据这些值进行操作。

onUpdateReceived = new BehaviorSubject<any>(null);

startContinuousUpdate(): void {
    this.updateSubscription = this.timerSub = timer(0, 1000).pipe(
        mergeMap(cycle => this.http.get("some url")
    ).subscribe((result: any) => {
        this.onUpdateReceived.next(result);
    })
}

叠加附记

正如本文前言中已经提到的。根据刚刚收到的状态，应在特定组件中触发一个或多个连接的叠加层。

在我解释所需行为之前，先简要说明一下叠加层本身的工作原理。覆盖是使用角度 CDK 覆盖功能实现的。 我的 OverlayService 被调用以打开基于某些配置的覆盖和一个组件，以使用组件门户在主覆盖组件中呈现。

open(config): OverlayRef {
    // do something with the config
    const ref = new OverlayRef(...);
    const injector = createInjector(ref, this.injector);
    ref.attach(new ComponentPortal(OverlayComponent, null, injector));
    return ref;
}

createInjector(ref: OverlayRef, i: Injector) {
    const tokens = new WeakMap([[OverlayRef, ref]]);
    return new PortalInjector(i, tokens);
}

模板化的 OverlayRef 类包含一个 Subject，因为它被注入到要在门户中呈现的最终组件，一旦用户使用一个值关闭覆盖，就会发出。

通常，这种结构已被证明可以很好地集成到基于 rx 的工作流程中。 但是，这种方法和计时器的使用似乎存在一个普遍问题。

实际问题的最小示例

实际问题发生在上述 BehaviorSubject 的订阅中，每 10 秒获取一个新值。

some-specific.compoenent.ts:

ngOnInit(): void {
     this.updateService.onUpdateReceived.pipe(
         switchMap(data => {

             if(this.overlaysOpen)
                 return of(null);

             return this.overlay.open(OverlayAConfig).afterClosed$.pipe(
                 mergeMap( choice => {
                     if(choice)
                         return this.overlay.open(YetAnotherConfig1).afterClosed$;
                         this.overlaysOpen = true;
                     else
                         return this.overlay.open(YetAnotherConfig2).afterClosed$;
                         this.overlaysOpen = true;
                 })
             }) 
     ).subscribe(result => {
         // do something with the final results being either null or a value
         doSomething(result);
         if(conditions_indicating_that_the_overlays_are_closed)
             this.overlaysOpen = false;
     })
}

这是实际代码的最小表示。

期望与现实（某种）

代码应该做什么：

如果有任何打开的覆盖：只需返回 null 并因此跳过。 如果没有打开覆盖：打开第一个覆盖，等待用户决定一个选项。就选择打开不同的叠加层，等待用户再次输入。只要应用程序正在运行，就继续这样做。在订阅块内：处理输入。

代码的作用：

一旦打开叠加层，用户就会阅读信息并决定下一步。 如果，这就是这里的问题，而第一个叠加层是开放的，无论用户做出什么选择，都会要求新的值将被“忽略”。发生这种情况（我认为）是因为订阅被覆盖（或删除，取消了您喜欢的名称）并且没有关于现在如何继续的信息，因为代码本身刚刚重新开始。

我一直在玩 switchMap（如示例中）与 mergeMap，但两者都没有任何区别。我可以（并且确实）在最终订阅中触发覆盖代码（这显然有效），但这是一种反模式。此外，由于现在必须扩展结构，这根本不再适合。甚至没有提到它是一个 rx 反模式。

所以如果有人理解我的问题： 有没有办法触发这一系列事件，而不必在用户做出选择后终止计时器并重新启动它（这不是一个真正的选项，因为无法终止请求的状态->其他仍然需要的事情处理）

万一没人理解： 我注定要失败 :D 但我似乎找不到更好的方式来描述这个问题。 我可能能够创建一个最小的工作示例来伪造所有上述依赖项，但由于 sprint 只剩下有限的时间，我并没有真正看到它。

干杯！

Answer 1

你给我们的东西有些奇怪。

---

1

在高层次上，一个不只是盲目地每 10 秒运行一次的 api 轮询系统可能会为您提供良好的服务。有时您希望轮询 API 端点，有时又不想。考虑到这一点的系统会更加强大。

---

2

这对我来说没有意义：

mergeMap( choice => {
  if(choice)
     this.overlay.open(YetAnotherConfig1).afterClosed$;
     this.overlaysOpen = true;
  else
    this.overlay.open(YetAnotherConfig2).afterClosed$;
    this.overlaysOpen = true;
})

这实际上是mergeMap(a -> undefined)

我预计会出现以下错误：

错误：
您在预期流的位置提供了“未定义”。您可以提供 Observable、Promise、Array 或 Iterable。

---

3

也许可以快速解决：

• 使用exhaustMap 而不是switchMap。 exhaustMap 将忽略源排放，而内部 observable 仍处于活动状态。这将阻止用户的选择被覆盖，但您将在用户做出选择时从 API 端点删除更新。

• 使用concatMap 而不是switchMap。 concatMap 将在内部可观察对象仍处于活动状态时缓冲源排放。这将阻止用户的选择被覆盖，但如果用户需要 1 分钟来完成覆盖，系统将不得不同时处理来自 API 端点的 6 次更新。因此，您需要一种在不破坏用户体验的情况下处理此问题的方法。