rxjs 根据响应中指定的持续时间重复 api 调用

Question

上下文

我需要从具有可变过期时间（在响应中指定）的服务器获取数据。一旦到期，我需要再次获得它。所以我想创建一个流，它发出一个异步请求并在该请求的响应中指定的时间后重复该请求。

我尝试了什么

这是我的第一次尝试，但 repeatWhen 无法访问最后的回复。而不是每 1000 毫秒重复一次，我想根据响应的过期属性来做。

const { Observable, defer, of } = rxjs;
const { repeatWhen, delay, map, take } = rxjs.operators;

let count = 0;
function api() {
  return of({ data: count++, expiration: Math.random() * 1000 });
}

defer(() => api()).pipe(
  repeatWhen((notification) => notification.pipe(delay(1000))),
  map((response) => response.data),
  take(5)
).subscribe((x) => { console.log(x); });

<script src="https://unpkg.com/rxjs@rc/bundles/rxjs.umd.min.js"></script>

问题

使用 rxjs，如何进行 api 调用并根据其最后响应延迟重复调用？

更新

从技术上讲，这可以满足我的要求，但它有点笨拙……所以我想要一个更好的解决方案。

const { Observable, defer, of, BehaviorSubject, timer } = rxjs;
const { repeatWhen, delay, map, take, tap, switchMap } = rxjs.operators;

let count = 0;
function api() {
  return of({ data: count++, expiration: Math.random() * 1000 });
}

const trigger = new BehaviorSubject(0);
trigger.pipe(
  switchMap((expiration) => timer(expiration)),
  switchMap(() => api().pipe(
    tap((response) => { trigger.next(response.expiration); })
  )),
  take(5)
).subscribe((x) => { console.log(x); });

<script src="https://unpkg.com/rxjs@rc/bundles/rxjs.umd.min.js"></script>

Answer 1

这可以使用.expand() 运算符轻松实现，该运算符用于递归目的。确切的解决方案只有几行：

api()
    .expand(({expiration}) => api().delay(expiration))
    .take(5)
    .subscribe(x=>console.log(x));

这里是JSBin。

Answer 2

我不确定我是否完全理解了你的问题，但是这样的问题呢

function api() {
  return of(Math.random() * 10000);
}

defer(() => api()).pipe(
    tap(delay => console.log('delay', delay)),
    switchMap(data => interval(data)),
    take(5)
).subscribe(console.log);

评论后更新答案

您已经根据 api 返回给您的内容进行重复，而不是每 1000 毫秒。如果你运行这段代码应该很清楚

let count = 0;
const expiration = Math.random() * 1000;

function api() {
    return of({ count, expiration});
}

defer(() => api()).pipe(
    tap(delay => console.log('delay', delay.expiration)),
    switchMap(data => interval(data.expiration).pipe(map(() => data))),
    map(data => ({expiration: data.expiration, count: count++})),
    take(5)
).subscribe(console.log);

第二次评论后的第二次更新

如果我现在了解您想要实现的目标，这应该会对您有所帮助

defer(() => api()).pipe(
    tap(data => console.log('I do something with this data', data)),
    switchMap(data => interval(data.expiration)),
    switchMap(() => api()),
    take(5)
).subscribe(data => console.log('I do something with this data', data));