rxjs 中第一次发射的 BufferCount(2, 1)

Question

我有一个 Observable：

1----2----3----3----2----1-|

我用

.bufferCount(2, 1)

输出是

--[1,2]-[2,3]-[3,3]-[3,2]-[2,1]-|

然后我链

.filter(twoEmissions => twoEmissions[0] !== twoEmissions[1])

这使得：

 --[1,2]-[2,3]-------[3,2]-[2,1]-|

然后我链

.map(twoEmissions => twoEmissions[1])

只留下最新的排放量，所以我的最终输出是：

-----2-----3------------2----1---|

我的问题是在这种情况下省略了第一次发射。 我尝试使用 buffer() insted of bufferCount() 和以下closureNotifier：

bufferClosingNotifier = sourceObservable$
  .scan((acc, val, index) => (index), 0)
  .filter((index: number) => index === 0 || index > 0 && index % 2 !== 0)

它发出第 1、3、5、7... 发射，但它们不重叠。

我怎样才能有 bufferCount(2, 1) 但无论如何都会发射源 observable 的第一次发射？

Answer 1

如果您的要求如下：

给定一个值流：

1----2----3----3----2----1-|

我只想要成功的唯一值，以便结束流是

1----2----3---------2----1-|

您可以使用.distinctUntilChanged() 来获得这种行为：

public distinctUntilChanged(compare: function): Observable source

返回一个 Observable，它发出源发出的所有项目 通过比较与前一项不同的 Observable。

如果提供了一个比较器函数，那么它将为每个 项目来测试是否应该发出该值。

如果没有提供比较器功能，则使用相等检查 默认。

Rx.Observable.from([1,2,3,3,2,1])
  .distinctUntilChanged()
  .subscribe(val => console.log(val))

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.5/Rx.js"></script>