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() 的问题（第 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>

Answer 2

您正在寻找.switchMap 运算符。

Rx.Observable.interval(1000)
.switchMap(() => Rx.Observable.ajax(someUrl))
.subscribe(val => console.log(val));