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”你说，你加了一个 throttle

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

“这不是我想要的!看看你错过的所有值(value)”，指的是你对所有正在传输的值(value)进行 throttle 。

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() 的想法。和 throttleTimte() ，但是我无法想出解决方案，然后我尝试了其他方法。

我想出的运算符是 throttleDistinct()如您所愿:StackBlit Editor Link

它有2个参数，分别是:

duration: number以毫秒为单位，类似于
持续时间 throttleTime(duration: number)

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),
      )
  }
}