ForkJoin 中的计时器

Question

我有一个 API 有 3 个 GET 调用。

获取环境
获取产品
获取特定环境中特定产品的状态

我想做的是每 5 分钟轮询每个环境中每个产品的状态，这样我就可以在 table 中显示数据并刷新。

获取环境和产品的调用应该执行一次，因为这是静态信息。

let envs = this.http.get('http://my-api/envs);
let products = this.http.get('http://my-api/products);

forkJoin([envs, products]).subscribe(results => {
  //the code below should be executed every 5 mins and store the results in array
  results[0].forEach(env => {
    results[1].forEach(product => {
      this.http.get('http://my-api/status/${env}/${product}')
    })
  })
})

如有任何帮助，我们将不胜感激。

Answer 1

不确定这会完成什么，但它每五分钟运行一次订阅中的代码。好吧，更准确地说，它每 5 分钟调用一次 forkJoin。如果 forkJoin 需要可变的时间来执行，可能会稍微改变时间。否则，您可以提前调用 forkJoin 并使用 delay 和 Date()

上的一些数学将结果延迟到五分钟标记

const fiveMinutes = 1000 * 60 * 5;
timer(0, fiveMinutes).pipe(
  switchMap(_ => forkJoin({
     envs: this.http.get('http://my-api/envs'),
     products: this.http.get('http://my-api/products')
  }))
).subscribe(({envs, products}) => {
  envs.forEach(env => {
    products.forEach(product => {
      // This http call doesn't do anything ?
      this.http.get('http://my-api/status/${env}/${product}') 
    })
  })
});

更新#1：最终调用的间隔。

这个人每 5 分钟调用一次您的 this.http.get('http://my-api/status/${env}/${product}') 调用并将结果作为数组发送到订阅。

由于显而易见的原因，这没有经过测试。但它应该提供一个很好的起点。

const fiveMinutes = 1000 * 60 * 5;
forkJoin({
  envs: this.http.get('http://my-api/envs'),
  products: this.http.get('http://my-api/products')
}).pipe(
  map(({env, products}) => 
    envs.map(env =>
      products.map(product =>
        this.http.get('http://my-api/status/${env}/${product}') 
      )
    ).flat()
  ),
  mergeMap(statusCalls => timer(0, fiveMinutes).pipe(
    mergeMap(_ => forkJoin(statusCalls))
  ))
).subscribe(statusCallsResults => {
  //** Update your view with results **//
})

如果您真的想要之前设置的订阅，可能如下所示：

const fiveMinutes = 1000 * 60 * 5;
forkJoin({
  envs: this.http.get('http://my-api/envs'),
  products: this.http.get('http://my-api/products')
}).pipe(
  mergeMap(res => timer(0, fiveMinutes).pipe(
    map(_ => res)
  ))
).subscribe(({envs, products}) => {
  // This is called every five minutes
  envs.forEach(env => {
    products.forEach(product => {
      // This http call doesn't do anything ?
      this.http.get('http://my-api/status/${env}/${product}') 
    })
  })
});

更新#2：错误处理

forkJoin 如果其任何内部可观察对象失败，则将失败。您可以通过多种方式处理此问题。最基本的是将出错的流转换为发出 null 并成功完成的流。然后 forkJoin 将在每个数组位置都有 null 并出现错误。

可能看起来像这样：

const fiveMinutes = 1000 * 60 * 5;
forkJoin({
  envs: this.http.get('http://my-api/envs'),
  products: this.http.get('http://my-api/products')
}).pipe(
  map(({env, products}) => 
    envs.map(env =>
      products.map(product =>
        this.http.get('http://my-api/status/${env}/${product}').pipe(
          catchError(err => of(null))
        )
      )
    ).flat()
  ),
  mergeMap(statusCalls => timer(0, fiveMinutes).pipe(
    mergeMap(_ => forkJoin(statusCalls))
  ))
).subscribe(statusCallsResults => {
  //** Update your view with results **//
})

错误代码 500 的描述性不强。您可能会查看重试运算符以在请求失败时重试该请求。可以结合 catchError.

首先不从服务器获取该错误取决于您。

ForkJoin 中的计时器

Timer in ForkJoin

rxjs

angular

rxjs6

更新#1：最终调用的间隔。

更新#2：错误处理