javascript - RxJs - 最后一次请求间隔

Question

我想知道如何在 RxJS 中获得类似下面的内容。

1. 每隔一个请求都会发送到服务器。
2. 响应可能需要不同的时间。 (超过 1s 用于澄清)
3. 但是当只有一些响应到达时，所有较早的未完成的请求都将被取消，并且订阅将只收到这个最新完成的请求的结果。

看起来 switchMap 不能正常工作。因为每个下一个开始的请求都会在响应之前取消前一个请求。请参阅下面的代码。

let counter = 0;
let times = [1200, 500];

let getSomePromise = function() {
  return new Promise((resolve) => {
    let i = counter;
    counter++;
    
    if (i < 2) {
      console.log('promise ' + i + ' start');
    }
    
    setTimeout(() => {
      if (i < 2) {    
        console.log('promise ' + i + ' finish');
        resolve(i);
      }
    }, times[i]);
  });
}


let ob = Rx.Observable.interval(1000)
  .switchMap(() => {
     return getSomePromise();
  });

let sub = ob.subscribe((value) => {
  console.log('result ' + value);
});

<script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"></script>

我得到:

promise 0 start
promise 1 start
promise 0 finish
promise 1 finish
result 1

我要:

promise 0 start
promise 1 start
promise 0 finish
result 0
promise 1 finish
result 1

谢谢。

Answer 1

您可以使用mergeMap() 运算符来获得您想要的结果。此运算符订阅从其投影函数返回的所有内部 Observables 并重新发出它们的所有项目。

然而，据我了解你的问题你真的想使用 switchMap() (第 n.3 点描述了 switchMap 的作用)但是你的预期输出看起来你想要 mergeMap()。

查看更新后的代码及其输出:

let counter = 0;
let times = [1200, 500];

let getSomePromise = function() {
  return new Promise((resolve) => {
    let i = counter;
    counter++;
    
    if (i < 2) {
      console.log('promise ' + i + ' start');
    }
    
    setTimeout(() => {
      if (i < 2) {    
        console.log('promise ' + i + ' finish');
        resolve(i);
      }
    }, times[i]);
  });
}


let ob = Rx.Observable.interval(1000)
  .mergeMap(() => {
     return getSomePromise();
  });

let sub = ob.subscribe((value) => {
  console.log('result ' + value);
});

<script src="https://unpkg.com/@reactivex/rxjs@5.0.3/dist/global/Rx.js"></script>