使用 rxjs 按阈值展平数值序列答案

【问题标题】：Flatten sequence of numeric values by threshold with rxjs使用 rxjs 按阈值展平数值序列
【发布时间】：2015-02-04 20:02:08
【问题描述】：

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

换句话说：序列中的第一个值总是被发出。然后，每个发出的（=未过滤的）值应该至少比之前发出的值大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 和响应式编程的新手，我认为上述解决方案过于复杂。更重要的是，它违反了响应式编程的原则，即不要在组合管道之外保存状态。但我这样做了，因为我正在通过 last 变量跟踪并希望摆脱它。

我该如何解决这个问题并以被动的方式做到这一点？

【问题讨论】：

标签： javascript reactive-programming rxjs

【解决方案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; });

【讨论】：

【解决方案2】：

我知道这个问题已经得到解答，但如果你发现自己经常这样做，你可以自己动手。只是因为它是一个有趣的实用程序而吐槽：

Observable.prototype.maxThreshold = function(threshold, selector) {
  var last = null;
  return this.filter(function(x) {
    var n = selector ? selector(x) : x;
    var delta = Math.abs(n - last);
    if(last === null || delta > threshold) {
      last = n;
      return true;
    }
    return false;
  });
});

可以如下使用：

streamOfNumbers.maxThreshold(0.5).
  subscribe(function(x) {
    console.log(x);
  });

或

streamOfObjects.maxThreshold(0.5, function(x) { return x.value; }).
  subscribe(function(x) {
    console.log(x);
  });

【讨论】：