RXJS - Typescript - 等待 observable 发射，直到另一个 observable<boolean> 为 false答案

【问题标题】：RXJS - Typescript - Await observable emit until another observable<boolean> is falseRXJS - Typescript - 等待 observable 发射，直到另一个 observable<boolean> 为 false
【发布时间】：2021-06-30 00:35:09
【问题描述】：

我目前有一个如下所示的解决方案：当用户在视图中导航时，将执行 api 调用以获取该特定视图的数据。该数据随后又用于确定在用户执行操作时是否应出现对话框。

目前该解决方案正在运行，但当用户具有高延迟并在 api 请求完成之前执行上述操作时，我可以想到一个潜在的问题。
我不想在每次用户导航时都显示加载指示器，因为在使用这样的系统时它会困扰我。我宁愿让用户在她/他想要执行操作时等待。
此外，我确实有一个 observable，它指示我计划以某种方式使用的 api 请求是否正在加载。
为了清楚起见，代码看起来像这样，只是对名称做了一些小改动。

actionPerformed$ = createEffect(() => 
      this.actions$.pipe(
        ofType(performAction),
        withLatestFrom(entities$),
        mergeMap(data => {
          let payload = data[0];
          let dataFromApi = data[1];
          ...
          ...
        })

想到一个听起来不太有效的想法是让一个操作员检查条件并在加载时抛出错误，然后在短暂延迟后重试。

你有更好的建议吗？我一直在看 delayWhen 但似乎没有像这样的延迟。我宁愿有一个“门”，它在 loading$ 设置为 false 时打开流，并且在满足该条件时立即发出该值。

提前致谢！

【问题讨论】：

你使用路由吗？
@Yuriy 是的，我确实使用路由。

标签： angular typescript rxjs ngrx

【解决方案1】：

在这个例子中，一旦source$ observable 发出了false，dependent$ observable 就会发出一个值。请查看工作代码：https://stackblitz.com/edit/rxjs-r3yma1?devtoolsheight=33&file=index.ts

import { of, timer } from "rxjs";
import { filter, map, switchMap, take, tap } from "rxjs/operators";

const arr = [true, true, true, true, true, true, false];
const source$ = timer(0, 1000).pipe(
  map(i => arr[i]),
  tap(console.log)
);

const dependent$ = of("the other guy said false");

source$
  .pipe(
    filter(val => val === false),
    switchMap(() => dependent$),
    take(1)
  )
  .subscribe(console.log);

【讨论】：

感谢您的回复！这是在 observable 发出多个值的前提下工作的。我正在使用的案例只会在用户执行操作时发出一个值。如果在这种情况下 loading$ 为真，则该过滤器将永远不会被再次评估。也许我想要实现的目标是任何现有的 RXJS 操作符都无法实现的，我必须构建某种自定义功能。
嗨@Tjulu。我发布了另一个答案。我希望这个更类似于您的用例。

【解决方案2】：

在此示例中，依赖的 observable 将立即发出，因为服务尚未从 API 获取数据。然后服务将开始获取数据，并且一旦服务完成，依赖的 observable 将发出。你可以在https://stackblitz.com/edit/rxjs-3ecq5d?devtoolsheight=33&file=index.ts看到工作代码

import { BehaviorSubject, of } from "rxjs";
import { filter, switchMap, take } from "rxjs/operators";

class Service {
  loading$ = new BehaviorSubject<boolean>(false); // Initialized false, this is the initial situation before you start fetching data from the API

  fetchData() {
    console.log("Fetching data...");
    this.loading$.next(true);
    setTimeout(() => {
      console.log("Finished fetching data");
      this.loading$.next(false);
    }, 5000);
  }
}

class Component {
  service: Service;

  constructor(service: Service) {
    this.service = service;
  }

  emitWhenReady() {
    const dependent$ = of("I'll emit when not fetching data");
    this.service.loading$
      .pipe(
        filter(loading => loading === false),
        switchMap(() => dependent$),
        take(1)
      )
      .subscribe(console.log);
  }
}

const service = new Service();
const component = new Component(service);
component.emitWhenReady(); // This guy will emit immediately because the service is not fetching data yet;
service.fetchData();
component.emitWhenReady(); // This guy will wait until the service has finished fetching data.

【讨论】：

【解决方案3】：

听起来你可以像这样使用combineLatest 而不是withLatestFrom somehwat：

combineLatest([
  this.actions$.pipe(ofType(performAction)),
  entities$
]).pipe(
  filter(([action, entities]) => entities != null), // condition depends on entities structure
  map(([action, entities]) => {
    // doStuff
  })
);

这将等待两个可观察对象至少发射一次，这将转换为加载的“实体”。过滤器只是为了安全。如果在 entitites$ 调用完成之前有一个用户操作，它还将考虑最后一个用户操作。

【讨论】：