RxJS 限制相同的值但让新值通过

Question

“给你”，有人说，你得到了这个输入的值流，你有点想对它做 distinctUntilChanged() ......

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

到目前为止没有什么奇怪的，
但是现在有人说如果相同的值再次出现“没关系”，“但前提是不会很快！”。我希望至少'----'在相同的值之间有刻度。“Okey”你说，你加了一个油门

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

首先，我想出了以某种方式组合distinctUntilChanged()and 的想法throttleTimte()，但是我无法想出解决方案，然后我尝试了其他方法。

我想出的运算符throttleDistinct()可以按照您的意愿工作：StackBlit Editor Link

它有2个参数，分别是：

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 2

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

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

之前的示例，在StackBlitz 示例中

更新：这实际上并不能 100% 工作。它只考虑当前窗口。所以你可以例如

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

其中[----]代表时间窗口。换句话说，如果一个值首先在一个窗口中最后发出，并在下一个窗口中首先发出，则相同的值将紧接着彼此通过。

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