javascript - 如何使用 Rxjs 5 限制服务器请求？

Question

我想创建一个函数来在服务器中发出 HTTP PUT 请求，但如果在该时间间隔内多次调用该函数，则每 500 毫秒使用最后一次调用参数仅发送一次请求，如果仍在进行中，则取消最后一次请求。

我研究并提出了这个解决方案:

const { Observable } = require('rxjs/Observable')
const { Subject } = require('rxjs/Subject')
const { switchMap, auditTime } = require('rxjs/operators')

// Simulate HTTP request
function makeRequest (val) {
  return Observable.create(observer => {
    console.log('Request:', val);
    observer.next(val);
    observer.complete();
  });
}

const toUpdateStream = new Subject();
const notifier$ = toUpdateStream.pipe(
  auditTime(500),
  switchMap(val => makeRequest(val))
);


function updateThrottle (val) {
  return Observable.create(observer => {
    const lastUpdate$ = notifier$.subscribe(res => {
      observer.next(res);
      observer.complete();
      lastUpdate$.unsubscribe();
    });
    toUpdateStream.next(val);
  });
}

// Try to update 3 times with different parameters
updateThrottle(10).subscribe(val => { console.log('1:', val); });
updateThrottle(20).subscribe(val => { console.log('2:', val); });
updateThrottle(30).subscribe(val => { console.log('3:', val); });

输出是:

Request: 30
1: 30
Request: 30
2: 30
Request: 30
3: 30

问题是我只需要用 30 调用请求一次，而不是每次都调用。

我能做什么？

Answer 1

因此，只有当值与上一次发射相比发生变化时，您才希望将值传递给 auditTime。 pairwise运算符非常适合此用例。

const notifier$ = toUpdateStream.pipe(
  startWith(null),
  pairwise(), // Emit the previous and current search options
  filter(([oldSearch, newSearch]) => oldSearch !== newSearch),
  map(([oldSearch, newSearch]) => newSearch),
  auditTime(500),
  switchMap(val => makeRequest(val))
);

除了 pairwise()，您还可以使用 bufferCount(2, 1)，它会做同样的事情。