rxjs中第一次发射的缓冲计数（2，1）

Question

我有一个可观察到的：

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---|

我的问题是在这种情况下省略了第一个发射。我尝试使用BufferCount（）的缓冲区（），并使用以下闭包通知程序：

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

它发射第一、第三、第五、第七......发射，但它们不重叠。

我怎么能有缓冲区计数（2,1），但发射源的第一次发射可观察到？

Answer 1

如果您的要求如下：

给定一系列值：

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

我只希望后续的唯一值，以便结束流

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

您可以使用。distinctUntilChanged（）要获取此行为：

返回一个可观察的，该可观察发出的源可观察发出的与前一项相比不同的所有项。

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

如果未提供比较器函数，则默认情况下使用相等检查。

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>