RXJS bufferedAmount / 累加值 / 复位减少答案

【问题标题】：RXJS bufferedAmount / accumulate value / reduce with resetRXJS bufferedAmount / 累加值 / 复位减少
【发布时间】：2020-08-11 12:05:56
【问题描述】：

我有一个价值观流

[ 0.5, 0.3, 0.4, 0.6, 1.4, 0.3, 0.6 ]

我想把它变成

[           1         2         1 ]

所以，我们正在累积第一个流的值，直到达到一个整数（至少为 1），然后在累积剩余量的同时发出整数。

这完全出乎我的意料，我认为 switchMap 的解决方案指日可待。

【问题讨论】：

标签： typescript rxjs rxjs6

【解决方案1】：

这是我的方法：

src$ = src$.pipe(publish());

const wholeNumber$ = src$.pipe(
  scan(
    (acc, crt) => (acc | 0) > 1 ? crt + (+(acc - (acc | 0)).toPrecision(1)) : acc + crt, 0
  ),
  map(v => (v | 0)),
  filter(v => v >= 1),
);

src$.pipe(
  buffer(wholeNumber$)
).subscribe();

publish 将确保源不会被多次订阅。它也是multicast(new Subject()) 的简短版本，基本上是一种多播源的方式。为了使其工作，src$ 必须异步发出，以便使用中的Subject 正确注册其订阅者（wholeNumber$ 和另一个）。

如果源不异步发出，您可以强制使用src$.pipe(observeOn(asapScheduler)) 这样做，这会将每个通知安排为一个承诺。

让我们仔细看看提供给scan的cb：

(acc, crt) => (acc | 0) > 1 ? crt + (+(acc - (acc | 0)).toPrecision(1)) : acc + crt`

number | 0 与Math.trunc(number) 相同。

在+(acc - (acc | 0)).toPrecision(1):

当你执行1.2 - 1时，你会得到：0.199...96;与toPrecision(1)：(1.2 - 1).toPrecision(1) = "0.2"。 + 将得到 0.2 作为数字。

【讨论】：

【解决方案2】：

谢谢安德烈，

我已使用您的代码并将其更改为自定义运算符。


export const bufferAmount = (
  amount: number
): MonoTypeOperatorFunction<number> => (
  source: Observable<number>
): Observable<number> =>
  new Observable<number>((observer) => {
    let bufferSum = 0;
    return source.subscribe({
      next(value) {
        bufferSum += value;
        if (bufferSum < amount) return;
        const nextSum = Math.trunc(bufferSum);
        bufferSum -= nextSum;
        observer.next(nextSum);
      },
      error(error) {
        observer.error(error);
      },
      complete() {
        observer.complete();
      },
    });
  });

ticker.pipe(bufferAmount(1)).subscribe();

【讨论】：