javascript - Rx.Js : understanding expand operator

Question

我将数据发布到后端，处理数据需要一些时间，并且在我的特定情况下长时间轮询不是解决方案，所以我每 5 秒发送一次请求 expand运算符(operator)

this.apiService.postData(data).pipe(
    expand((status) =>
        status.comptete? this.apiService.askStatus(status.request_id).pipe(delay(5000)) : empty()
    ),
    map((result) => {
        // processing result here
    })
);

问题是如何使延迟成为动态的(例如，第一次我想在 1 秒内询问状态，第二次在 2 秒内询问状态，依此类推)？还有两个问题。我是否正确理解，如果我添加 take(N)将限制 askStatus 的运算符调用 N ?我是否正确理解我不需要在这里取消订阅？

Answer 1

expand()每次调用项目函数时也会传递索引，因此您可以基于此计算延迟:

expand((status, index) => 
  status.comptete ? this.apiService.askStatus(...).pipe(delay(...)) : empty()

使用 take(N)内 expand()不会有帮助，因为 expand()对来自源和内部 Observable 的每次发射调用项目函数。但是你当然可以使用 take(N)之后 expand() .
您不必取消订阅 askStatus()如果您稍后在您还订阅的地方处理取消订阅，则手动。