3

我是 ReactiveExtensions 的新手，我没有开始工作。我认为应该是一个非常常见的用例。我只想在经过特定时间段而没有新的下一个值的情况下接收新值。在下面的示例中，此时间段为 1 秒。debounce -operator似乎完全符合我的要求。我没有让它工作。

const observable$ = new Rx.Observable(observer => {

 observer.next('start');

 setTimeout(() => {
  observer.next(1);
 }, 100);
 setTimeout(() => {
  observer.next(2);
 }, 200);
 setTimeout(() => {
  observer.next(3);
 }, 300);
 setTimeout(() => {
  observer.next(4);
 }, 400);

 setTimeout(() => {
  observer.next('end');
 }, 1500);
});

let sub = observable$
      .debounce(1000) //debounce(1000, null) does not work either
      .take(100)
      .subscribe(data => {
          console.log(data);
         }, 
         err => console.log(err.message), 
         complete => console.log('Observable completed')
       )

我想要得到的只是一个控制台输出：

"start"
"end"

在我的 IDE (Webstorm) 中，尽管文档声明第二个参数是可选的，但上面的代码甚至无法编译。在jsbin.com 上，我收到以下错误：“this.durationSelector.call 不是函数”（我承认，我还不知道如何在 rxjs 中应用调度程序）。在文档中，他们也只使用一个数字。我在 google 上找到的大多数 debounce 示例仅使用一个数字，即Stackoverflow 上的这个示例。为什么这在我的情况下不起作用？

谢谢你的帮助！

PS：我使用 rxjs 5.0.0-beta.6。

编辑：在这里的答案的帮助下，我确实找到了我想要的实际解决方案：

const observable$ = new Rx.Observable(observer => {

observer.next('start');

 setTimeout(() => {
  observer.next(1);
 }, 1100); //<-- If you change 1100 to i.e. 900 you just get "end" in the output, because there is no 1s periode during which no new value arrives. 
 setTimeout(() => {
  observer.next(2);
 }, 1200);
 setTimeout(() => {
  observer.next(3);
 }, 1300);
 setTimeout(() => {
  observer.next(4);
 }, 1400);
 setTimeout(() => {
  observer.next(5);
 }, 1500);

 setTimeout(() => {
  observer.next('end');
 }, 1501);


});

let sub = observable$
      .debounceTime(1000)
      .take(10)
      .subscribe(data => {
         console.log(data);
       }, 
       err => console.log(err.message), 
       complete => console.log('Observable completed')
);
4

2 回答 2

4

这只是针对旧版本 RxJS 的文档的问题。RxJS 4重载debounce 在这里。

但是，RxJS 5将运算符拆分为实际上有两个，一个使用选择器函数来确定去抖动长度，第二个debounceTime接受时间。

因此，您的流变为：

let sub = observable$
      .debounceTime(1000) //debounce(1000, null) does not work either
      .take(100)
      .subscribe(data => {
          console.log(data);
         }, 
         err => console.log(err.message), 
         complete => console.log('Observable completed')
       )
于 2016-09-07T03:44:31.423 回答
2

你可以做这样的事情

const observable$ = new Rx.Observable(observer => {

    observer.next('start');

    setTimeout(() => {
        observer.next(1);
    }, 100);
    setTimeout(() => {
        observer.next(2);
    }, 200);
    setTimeout(() => {
        observer.next(3);
    }, 300);
    setTimeout(() => {
        observer.next(4);
    }, 400);

    setTimeout(() => {
        observer.next('end');
    }, 1500);


});

let sub = observable$
    .map(function(x, i) {
        return {
            val: x,
            index: i
        };
    })
    .debounce(function(obj) {
        let interval = obj.index === 0 ? 0 : 1500;
        return Rx.Observable.timer(interval);
    })
    .take(100)
    .subscribe(data => {
            console.log(data.val);
        },
        err => console.log(err.message),
        complete => console.log('Observable completed')
    )

这里的关键是使用map函数获取元素的索引，然后决定等待间隔。

于 2016-09-06T22:16:18.307 回答