javascript - 使用 rxjs 按阈值展平数值序列

Question

使用 rxjs，我得到了一个可观察的 float 序列。现在我想过滤掉流中较小的变化，并且只在它比以前发出的值大一定数量时发出一个值。

换句话说:序列中的第一个值总是被发出。然后，每个发出的(=未过滤的)值应该至少比之前发出的值大 delta。任何不符合该条件的值都会被过滤掉。

我已经想出了一个解决方案，可以按照我上面的描述进行操作:

var obs = Rx.Observable.create(function(observer) {
  /* ... */
});

var last;

obs.map(function(value) {
  if (last === undefined) {
    last = value;
    return value;
  } else {
      var threshold = 0.5,
          delta = Math.abs(last - value);

      if (delta > threshold) {
        last = value;
        return value;
      }
      else {
        return undefined;
      }
    }
  }).
  filter(function(value) {
    return value !== undefined;
  });

我是 rxjs 和响应式编程的新手，我认为上述解决方案过于复杂。更重要的是，它违反了响应式(Reactive)编程的原则，即不要在组合管道之外保存状态。但我这样做了，因为我通过 last 变量跟踪并希望摆脱它。

我该如何解决这个问题并以被动方式解决？

Answer 1

您可以使用scan 来管理您的状态:

var filtered = obs.scan({}, function (acc, value) {
    if (acc.value !== undefined) {
        var threshold = 0.5,
            change = Math.abs(acc.value - value);

        if (change < threshold) {
            return { value: acc.value, isValid: false };
        }
    }

    return { value: value, isValid: true };
})
.filter(function (acc) { return acc.isValid; })
.map(function (acc) { return acc.value; });