Angular/RxJS - 有一个 RxJS 管道用于流启动？答案

【问题标题】：Angular/RxJS - There is a RxJS pipe for stream started?Angular/RxJS - 有一个 RxJS 管道用于流启动？
【发布时间】：2020-06-18 11:39:28
【问题描述】：

我正在尝试创建一种在解析某些流时动态呈现和隐藏加载屏幕的方法。

这是我当前的代码：

this.requestService
        .get<Checkout>(`check/${orderNumber}`)
        .pipe(
            tap(() => this.startLoading()),  //i think this isnt the right way to use this pipe
            finalize(() => this.endLoading())
        )
        .subscribe(
            data => {
                data.orderNumber = orderNumber
                this.checkout.next(data)
            },
            error => {
                this.notification.warning(error)
            }
        )

预期的结果是当我的流开始时，以startLoading() 显示加载屏幕，在操作完成时结束，使用endLoading() 隐藏加载。

我的工作代码：

this.startLoading() //basically calling function before i create the stream

this.requestService
        .get<Checkout>(`check/${orderNumber}`)
        .pipe(                
            finalize(() => this.endLoading())
        )
        .subscribe(
            data => {
                data.orderNumber = orderNumber
                this.checkout.next(data)
            },
            error => {
                this.notification.warning(error)
            }
        )

我正确使用了这个tap 管道吗？是否有其他管道可以更好地解决此问题？

使用 RxJS 最好的方法是什么？

【问题讨论】：

使用第一个示例我的startLoading() 函数在流启动时不会被调用
您希望在全局范围内还是在一个组件中执行此操作？
在角度服务中，但我不知道它是否重要

标签： angular typescript rxjs stream reactive-programming

【解决方案1】：

在您的第一个示例中，您的点击在您的 http 请求完成后运行。

最终，您只需在启动 http 请求之前调用 this.startLoading()。

this.startLoading();
this.requestService.get<Checkout>(`check/${orderNumber}`).pipe(
  finalize(() => this.endLoading())
).subscribe(() => {

});

如果你真的想在管道中调用this.startLoading()，你可以在 http 请求开始之前调用它，从你自己的 observable 开始：

return of(null).pipe(
  tap(() => this.startLoading()),
  concatMap(() => this.requestService.get<Checkout>(`check/${orderNumber}`)),
  finalize(() => this.endLoading())
).subscribe(() => {

});

但是这样做并没有多大意义。

所以您的tap 语法是正确的，只是在您认为应该执行的时候没有执行。

【讨论】：

【解决方案2】：

如果您想对每个可观察订阅执行副作用，您可以使用 defer() 扭曲源。

import { defer } from 'rxjs';

...

defer(() => {
  this.startLoading();
  return this.requestService.get<Checkout>(`check/${orderNumber}`);
}).pipe(
    finalize(() => this.endLoading())
  )
  .subscribe(...);

【讨论】：

【解决方案3】：

您可以使用 BehaviorSubject 和 RxJs defer 和 finalize 管道来跟踪加载过程。

import { defer } from 'rxjs';

// ...

public this.loading: BehaviorSubject<boolean> = new BehaviorSubject(false);

// ...

public getOrder(orderNumber): Observable<Checkout> {
  defer(() => {
    this.loading.next(true);
    return this.requestService.get<Checkout>(`check/${orderNumber}`);
  }).pipe(
    finalize(() => this.loading.next(false)
  );
}

// to get boolean from BehaviorSubject -> is loading
public isLoading(): boolean {
  return this.loading.value;
}

// ...

您现在还可以在您的类之外使用this.loading observable，例如根据可观察值显示或隐藏加载器：

// logs to console each time loading state changes
serviceName.loading.subscribe(result => console.log(result));

以后不要忘记取消订阅，因为loader是一个行为主体，它是多播的，如果不取消订阅会导致内存泄漏。

【讨论】：