如何在动态数组中设置订阅

debugcn 发表于 Dev

数字污染

我正处于学习 Angular 的早期阶段，正在尝试了解 RxJS、Observable 等。可能我在语法和概念上都有问题。

所以这里我有一个服务incidentService，它获取一系列事件并将它们显示在页面上。如果该incident.requirementMet属性是，true那么我希望该incident.resolved属性在列表中每个事件的计时器上更新。导航到另一个视图时，我想取消所有订阅。

这是我迄今为止尝试过的。setIncidentSubscriptons()和postIncident()方法周围出现问题。

Angular v8.2 rxjs v6.4

import { Component, OnInit, OnDestroy } from '@angular/core';
import { Subscription, Observable, timer } from 'rxjs';
import { IncidentService } from '../services/incident.service';
import { Incident } from '../models/incident';

@Component({
  selector: 'app-incident',
  templateUrl: './incident.component.html',
  styleUrls: ['./incident.component.scss']
})
export class IncidentComponent implements OnInit, OnDestroy {
  private incidentSubscriptions: Subscription[] = [];
  incidents: Incident[];

  constructor(private incidentService: IncidentService) { }

  ngOnInit() {
    this.getIncidents();
  }

  ngOnDestroy() {
    this.incidentSubscriptions.forEach(subscription => subscription.unsubscribe());
  }

  getIncidents(): void {
    this.incidentService.getIncidents()
      .subscribe((incidents) => {
        this.incidents = incidents;
        this.setIncidentSubscriptons();
      });
  }

  setIncidentSubscriptons(): void {
    let timerDelay = 1000;

    for (const i in this.incidents) {
      if (this.incidents[i].requirementMet) {
        timerDelay += 2000;

        this.incidentSubscriptions.push(
          timer(1)
          .subscribe(
            () => { this.postIncident(this.incidents[i], timerDelay); }
          )
        );
      }
    }
  }

  postIncident(incident: Incident, timerDelay: number) {
    if (incident.resolved < 100) {
      setTimeout(() => {

        incident.resolved += 1;
        this.incidentService.updateIncident(incident).subscribe(() => {
          this.postIncident(incident, timerDelay);
        });

      }, timerDelay);
    }
  }

}

弗里多

你应该尝试用 RxJS 操作符来完成你所有的逻辑，并构建一个 Observable 来完成你所有的任务。只订阅一次最终的 Observable。

我假设this.incidentService.updateIncident(incident)只发出一次然后完成并且不使用发出的值。

import { timer, forkJoin, Observable, Subject } from 'rxjs';
import { tap, concatMap, switchMap, takeUntil, take } from 'rxjs/operators';

private onDestroy$ = new Subject();

ngOnInit() {
  this.getAndUpdateIncidents().subscribe();
}

ngOnDestroy() {
  this.onDestroy$.next();
  this.onDestroy$.complete();
}

getAndUpdateIncidents(): Observable<any[]> { // 'any' is whatever this.incidentService.updateIncident(incident) returns
  return this.incidentService.getIncidents().pipe(
    tap(incidents => this.incidents = incidents), // assign the returned incidents
    switchMap(incidents => this.updateIncidents(incidents)), // switch to an Observable that updates the incidents
    takeUntil(this.onDestroy$) // unsubscribe on destroy
  )
}

updateIncidents(incidents: Incident[]): Observable<any[]> {
  let timerDelay = 1000;
  return forkJoin(incidents // use forkJoin to execute an array of Observables parallely
    .filter(incident => incident.requirementMet) // only take incidents who meet the requirement
    .map(incident => { // map to an Observable that periodically updates the incident
      timerDelay += 2000;
      timer(0, timerDelay).pipe( // timer will emit numbers with a given delay
        take(100), // only take 100 numbers
        concatMap(_ => { // map to the Observable that updates the incident, 
          // concatMap ensures that the next update will only be executed when the previous completed
          incident.resolved += 1;
          return this.incidentService.updateIncident(incident);
        })
      )
    }));
}

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。