具有实时缓冲区输出的 RXJS 计数器答案

【问题标题】：RXJS Counter with live buffer output具有实时缓冲区输出的 RXJS 计数器
【发布时间】：2019-10-29 18:55:35
【问题描述】：

我正在尝试用 RXJS 实现一种计数器。有一个可以逐步增加或减少的主题（数字）。最后，它应该发出计数器值，可以是正数也可以是负数。

如下面的示例所示，输入会以一秒的时间去抖动。这意味着，observable 在不活动一秒后发出并返回值的总和。

const mySubject$: Subject<number> = new Subject<number>;

get counter$(): Observable<number> {
  return this.mySubject$.pipe(
    buffer(this.mySubject$.pipe(debounceTime(1000))),
    map(steps => steps.reduce((accumulator, currentValue) => accumulator + currentValue, 0))
  );
}

increment(): void {
  this.mySubject$.next(1);
}

reduce(): void {
  this.mySubject$.next(-1);
}

到目前为止很好......但是现在，我想在增量/减少的某个地方实时显示当前计数器值，然后在一秒钟后发出。有没有办法在去抖时间结束之前访问缓冲数组？

我可以在可观察对象之外创建另一个计数器变量，并将该值放在缓冲区（）之前的 tap() 中。但这似乎不是RXJS的方式？

const mySubject$: Subject<number> = new Subject<number>;
const counter: number = 0;

get counter$(): Observable<number> {
  return this.mySubject$.pipe(
    tap(value => this.counter += value),
    buffer(this.mySubject$.pipe(debounceTime(1000))),
    map(steps => steps.reduce((accumulator, currentValue) => accumulator + currentValue, 0)),
    tap(() => this.counter = 0)
  );
}

任何关于干净方式的想法，如何做到这一点？

代码的目的：我有一个在电视上运行的应用程序。在流式传输期间，用户可以使用遥控器上的两个专用键以 30 秒的步长向后或向前“跳跃搜索”。例如，如果用户连续按下按钮 3 次，则流将向后或向前跳跃 90 秒。按键动作之间有一秒钟的延迟，在累积值应该发出之前。但每次用户按下按钮时，都会有一个可见的输出（如祝酒词），说明跳转搜索将在多少秒内发出。

【问题讨论】：

我认为 debounceTime 并没有按照您的预期工作。 debounceTime(1000) 只会忽略间隔小于 1 秒的发射值。感觉您正在寻找的是一个计时器，但我不太了解您的代码的目的。您能否详细说明或发布使用示例？
结合缓冲区，debounceTime(1000) 正是我想要的。它“存储”所有值，直到一秒钟内没有其他增加/减少操作。我将使用有关使用情况的更多信息来编辑我的问题。

标签： angular rxjs buffer counter

【解决方案1】：

您也许可以尝试使用扫描，它会为您完成求和工作，因此您可以在某处提示它并仍然对输出进行去抖动。

get counter$(): Observable<number> {
  return this.mySubject$.pipe(
    scan((acc, delta) => delta ? acc + delta : 0, 0),
    tap((value) => *you can use the total accumulator so far here*),
    debounceTime(1000),
  );
}

要重置计数器，您只需发送一个空值。

【讨论】：

【解决方案2】：

我将使用一个主题 (_jumpBy$)，用于触发命令，以及两个映射的 observable：一个 (jumpByToShow$) 立即显示通过 scan 累积的值，另一个消除反射的发射首先，在发出一个值时，将累积的值清零，使主体发出一个-amount 值。

未经测试，希望对您有所帮助。

  private _jumpBy$ = new Subject<number>();

  // the quantity changing immediately, to be shown on screen
  public jumpByToShow$: Observable<number>;

  // the quantity debounced
  public jumpByToExecute$: Observable<number>;

  constructor(/* ... */) {
    this.jumpByToShow$ = this._jumpBy$.pipe(
      startWith(0),
      scan((amount, newAmount) => (amount + newAmount), 0),
      share()
    );

    this.jumpByToExecute$ = this.jumpByToShow$.pipe(
      debounceTime(1000),
      filter(amount => amount !== 0),
      tap(amount => this._jumpBy$.next(-amount))
    );
  }

【讨论】：