使用异步管道轮询 Observables？

Posted 2023-03-13

【中文标题】使用异步管道轮询 Observables？

【英文标题】：Use async pipe for polling Observables?

【发布时间】：2017-09-09 08:46:23

【问题描述】：

场景：

我有一个每 2 秒轮询一次 URL 的服务：

export class FooDataService 

...

  public provideFooData() 
    const interval = Observable.interval(2000).startWith(0);
    return interval
      .switchMap(() => this.http.get(this.requestUrl))
      .map(fooData => fooData.json())
  

现在我有一个组件，我想在其中显示这个轮询数据：

export class FooComponent implements OnInit 

  constructor(private fooDataService: FooDataService) 

  public fooData$: Observable<FooData>;

  ngOnInit() 
    this.fooData$ = this.fooDataService.provideFooData();
  

在组件模板中，我将使用异步管道来检索值并将其传递给子组件：

<foo-data-viewer [data]="fooData$ | async"></foo-data-viewer>

这种方法的问题：

这样实现它不能用量角器测试（见我的last question on this topic 或this article）。所有代码都在 ngZone 内执行，量角器将等待所有排队的操作完成后再继续。但是Observable.interval() 将排队无限数量的动作，从而导致量角器超时。

常见的解决方案：

我最常阅读的解决方法是像这样使用runOutsideAngular：

export class FooComponent implements OnInit, OnDestroy 

  constructor(private ngZone: NgZone,
              private fooDataService: FooDataService) 

  public fooData: FooData;
  public fooDataSubscription: Subscription<FooData>;

  ngOnInit() 

    this.ngZone.runOutsideAngular(() => 
      this.fooDataSubscription =
        this.fooDataService.provideFooData()
            .subscribe(
               fooData => this.ngZone.run(() => this.fooData = fooData)
            );
    );
  

  ngOnDestroy(): void 
    this.fooDataSubscription.unsubscribe();
  

通过在 ngZone 之外运行间隔，量角器在继续之前不会等待轮询完成，因此不会超时。

这意味着：

我无法使用async 管道，我必须手动订阅。 我必须手动管理订阅，并在组件被销毁时自行清理。 我无法保持漂亮干净的 Observable 样式，而且代码变得更加复杂，尤其是在我有多个轮询服务的情况下。

我的问题：

有没有办法在 ngZone 之外运行间隔时保持功能性 rxjs 样式并继续使用异步管道（或等效管道）？

我偶然发现了this github project，它看起来正是我想要的，但我无法让它工作。

我需要的是一个工作示例，可以像在我的场景中那样离开和重新进入区域，而无需自己管理订阅。

【问题讨论】：

有人写了一个不错的服务，这似乎可以解决问题：github.com/angular/protractor/issues/…

【参考方案1】：

import  Injectable  from '@angular/core';
import  interval, of, timer  from 'rxjs';
import  startWith, switchMap, delay, map, share, shareReplay  from 'rxjs/operators';
import  HttpClient  from '@angular/common/http';

@Injectable()
export class DataService 
  private fakeRequest$ = timer(1000).pipe(map(() => new Date().getTime()))
  private pollingData$ = interval(2000)
    .pipe(
      switchMap((index) => this.fakeRequest$),
      shareReplay(1)
    )
  constructor()  

  public getData$() 
    return this.pollingData$;
  

Stackblitz