rxjs缓冲区不像我期望的那样使用节流阀。

Question

我正试图在节流排放的基础上创建缓冲排放。下面的工作与RxJs 2中所期望的一样

.buffer(function() { return clickStream.throttle(250); })

这里是完整的jsfiddle示例

但是，当我试图在本地项目中使用它时，它会返回以下错误：

Uncaught :您提供了“() => clickStream.throttle(250)”，其中期望有一个流。您可以提供一个可观察的、许诺的、数组的或可迭代的。

这是我的密码

import Rx from 'rxjs/Rx';

const $button = document.getElementById('button');
const $label = document.getElementById('label');

const clickStream = Rx.Observable.fromEvent($button, 'click');

const doubleClickStream = clickStream
  .buffer(() => clickStream.throttle(250))
  .map(arr => arr.length)
  .filter(len => len === 2)

clickStream.subscribe(x=>console.log(x));

doubleClickStream
  .subscribe(() => $label.textContent = 'double click')
//
doubleClickStream
  .throttle(1000)
  .subscribe(() => $label.textContent = '-')

我知道我目前对这个库的了解还不够，也许它与rxjs版本有关(在工作示例中是2.x，在我上面的代码中是5.x)。

我还试图把它写成：

.buffer(clickStream.throttle(250))

它在单击按钮后引发另一个异常：

Uncaught : this.durationSelector不是函数

你能帮我理解一下这不对吗？

Answer 1

您的问题在于，RxJs5是一个完整的重写，并且有许多破坏性的更改。其中许多是操作符的重命名或在不同操作符中分割操作符重载。

您的错误是由.throttle()引发的，在RxJs5中，该函数将在缓冲区关闭时返回一个可观察到的函数。您要寻找的重载已在RxJs5中重命名为.throttleTime()。

为了这个原因，最好不要混淆不同版本的Rx。如果您打算开始使用RxJs5，请查看迁移指南，让您了解所做的事情。

Answer 2

这就是我如何在RxJs5.5中实现的

import { buffer, throttleTime, map, filter } from "rxjs/operators";

const clickStream = Observable.fromEvent(button$, 'click');
this.subscriptions.add(clickStream
  .pipe(
      buffer(clickStream.pipe(throttleTime(500))),
      map(arr => arr.length),
      filter(len => len === 4)
    )
  .subscribe(resp => {
    console.log('clicked', resp);
}))