RxJS：计算集合项的更改

Question

给定一组项目，在任何项目/项目属性的特定时间间隔内跟踪/监控/计数变化的最佳方法是什么？
更具体地说，我需要能够监控变化并通知特定的动态条件。

到目前为止我的解决方案：
每次发生变化时，我都会在 ReplaySubject 中推送数据更改日志。
我订阅了这个主题并应用了一些自定义运算符，这些运算符过滤特定列并计算更改。

例如。 '如果项目 Y 的属性 X 在过去 5 秒内更改了 2 次，请通知我'

// const dataChangeLog$ = new ReplaySubject(...)

    dataChangeLog$.pipe(
      filterColumnChanges('columnName'), // filters only changes on the given column
      windowTime(5000),
      switchMap((change$) => change$.pipe(countCellChanges(2)))
    );

    const countCellChanges = (
      limit: number,
      reset = true
    ): MonoTypeOperatorFunction<DataChangedLogEntry> => {
      return (source$) =>
        defer(() => {
          const counterMap = new Map<unknown, number>();

          const getCellCounter = (dataChangeLog: DataChangedLogEntry) =>
            counterMap.get(dataChangeLog.primaryKeyValue) ?? 0;

          return source$.pipe(
            tap((dataChangeLog) => {
              let currentCounter = getCellCounter(dataChangeLog);
              counterMap.set(dataChangeLog.primaryKeyValue, ++currentCounter);
            }),
            filter((dataChangeLog) => {
              return getCellCounter(dataChangeLog) >= limit;
            }),
            // if limit is reached, reset it
            tap((dataChangeLog) => {
              if (reset) {
                counterMap.set(dataChangeLog.primaryKeyValue, 0);
              }
            })
          );
        });
    };

问题是计数器每 5 秒重置一次，无论我在此间隔内进行了多少更改。
我真正需要的是检查每一次排放是否在最后 Y 秒内至少有 X 次排放。

Answer 1

此操作符检查源并在 您描述的条件满足。

const xInLastY = (x, y) => (source$) => {
  const runningCount$ = source$.pipe(
    mergeMap(() => {
      const inc$ = of(1);
      const dec$ = of(-1).pipe(delay(y));
      return of(inc$, dec$).pipe(mergeAll());
    }),
    scan((acc, next) => acc + next, 0),
  );

  return runningCount$.pipe(
    map((count) => count >= x),
    distinctUntilChanged(),
    filter(x => x)
  );
};

编辑：专门解决这个位：

如果项目 Y 的属性 X 在过去 5 秒内更改了 2 次，请通知我

运算符巧妙地融入了这个用例。

const recentChangesToProperty$ = source$.pipe(
  distinctUntilKeyChanged(MY_PROPERTY),
  xInLastY(2, 5e3)
);

最后更新：

这概括了“最后 X 毫秒的条件”的情况，以满足 OP 的其他要求：

仅当更改的值是最后 x 毫秒内的最高/最低时才发出

我们不是累积一个增长/缩小的计数，而是累积一个增长/缩小的值数组。 “max”的情况在下面实现，“min”的实现很容易从这里得到。

type ArrayAction<T> = { type: 'PUSH'; value: T } | { type: 'POP' };

const arrayReducer = <T>(state: T[], action: ArrayAction<T>) => {
  switch (action.type) {
    case 'PUSH':
      return [...state, action.value];
    case 'POP':
      return state.slice(1);
    default:
      return state;
  }
};

const timeTrailingValues = (ms: number) => <T>(
  source$: Observable<T>
): Observable<T[]> => {
  const runningList$ = source$.pipe(
    mergeMap((value) => {
      const in$ = of({ type: 'PUSH', value });
      const out$ = of({ type: 'POP' }).pipe(delay(y));
      return of(in$, out$).pipe(mergeAll());
    }),
    scan(arrayReducer, [])
  );
};

const maxInLastX = (ms: number) => <T>(
  source: Observable<T>
): Observable<boolean> =>
  source$.pipe(
    timeTrailingValues(ms),
    map((xs) => {
      const [latestValue] = xs.slice(-1);
      const rest = xs.slice(0, -1);
      return rest.every((value) => value <= latestValue);
    }),
    filter(isMax => isMax)
  );

// Still want to track the running count? Combine timeTrailingValues with this
const atSize = (x: number) => <T>(
  source$: Observable<T[]>
): Observable<boolean> => {
  return source$.pipe(
    map((xs) => xs.length >= x),
    distinctUntilChanged(),
    filter((x) => x)
  );
};