RxJS节流相同的值，但让新的值通过

Question

“这里有”，有人说，您得到了这个输入的值流，您有些想要对其执行distinctUntilChanged() .

Input:  '1-1----11---2--1122----1---2---2-2-1-2---|'
Output: '1-----------2--1-2-----1---2-------1-2---|'

到目前为止没什么奇怪的，

但现在有人说，“如果同样的价值再次出现，那就没问题”，“但前提是不要太快！”我希望至少'----'在相同的值之间运行。“好吧”你说，然后加了一个节流阀

const source = new Subject<number>();

// mysterious cave troll is randomly source.next(oneOrTwo)

const example = source.pipe(throttle(val => interval(4000)));

Input:  '1-1----11---2--1122----1---2---2-2-1-2-----|'
Output: '1------1----2----2-----1-------2-----2-----|'

“这不是我想要的！看看你错过的所有价值”，指的是你对所有被流的价值的节流。

Input:  '1-1----11---2--1122----1---2---2-2-1-2-----|'
Output: '1------1----2----2-----1-------2-----2-----|'
        '-------------->1<--------->2<----->1<------|' <-- Missed values

“给你看，让我给你看”那个神秘的人说，然后给你这个。

想要输出

Input:  '1-1----11---2--1112----1---2---2-2-1-2-----|'
Output: '1------1----2--1--2----1---2-----2-1-------|'

我对此的回答是，这感觉就像一个组合的窗口是行不通的。

从更有经验的人那里，

这是一个很难解决的问题吗？(或者我错过了一个明显的解决方案)

Answer 1

我找到了一个可行的解决方案，有人对此有什么看法吗？

source.pipe(
   windowTime(4000),
   concatMap(obs => obs.pipe(distinct()))
);

以前的示例，在StackBlitz实例中

更新：--它实际上不能100%工作。它只考虑到当前的窗口。所以你可以比如说

`[1-12][2---]` which would give `1--22---|`

其中[----]将表示时间窗口。换句话说，如果一个值首先在一个窗口中发出，然后在下一个窗口中首先发出，则相同的值将在另一个窗口后面传递。

谢谢@eric99让我意识到这一点。

Answer 2

首先，我想出了将distinctUntilChanged()和throttleTimte()结合在一起的想法，但是我不可能想出解决方案，然后我尝试了其他的方法。

我想出的操作符是throttleDistinct() ，它可以按您的意愿工作： StackBlit编辑器链接

它有两个参数：

1. duration: number，以毫秒为单位，类似于throttleTime(duration: number)中的持续时间
2. equals: (a: T, b: T) => boolean，它是用于比较前一项是否等于下一项的函数，该项具有(a, b) => a === b的默认实现

import { of, fromEvent, interval, Observable } from 'rxjs';
import { map, scan, filter, } from 'rxjs/operators';

const source = fromEvent(document, 'keypress')
  .pipe(map((x: any) => x.keyCode as number))

source
  .pipe(
    throttleDistinct(1000),
  )
  .subscribe((x) => console.log('__subscribe__', x));

export function throttleDistinct<T>(
  duration: number,
  equals: (a: T, b: T) => boolean = (a, b) => a === b
) {
  return (source: Observable<T>) => {
    return source
      .pipe(
        map((x) => {
          const obj = { val: x, time: Date.now(), keep: true };
          return obj;
        }),
        scan((acc, cur) => {
          const diff = cur.time - acc.time;

          const isSame = equals(acc.val, cur.val)
          return diff > duration || (diff < duration && !isSame)
            ? { ...cur, keep: true }
            : { ...acc, keep: false };
        }),
        filter((x) => x.keep),
        map((x) => x.val),
      )
  }
}

Answer 3

从我的头顶上，你想要缓冲的时间间隔，然后在每个缓冲区内不同。

实际上，您希望每n毫秒重新启动一次不同的运行。

source.pipe(
  bufferTime(ms),
  mergeMap(bufferArray => from(bufferArray).pipe(distinctUntilChanged()) )
)