RxJS - 使重置无状态的计数器？

Question

假设我有以下标记:

<button id="dec">-</button>
<output id="out">0</output>
<button id="inc">+</button>
<button id="res">RESET</button>

以下Rx.js脚本:

var total = 0

Rx.Observable.merge(
    // decrement
    Rx.Observable.fromEvent($('#dec'), 'click')
    .map(function() { return -1 }),

    // increment
    Rx.Observable.fromEvent($('#inc'), 'click')
    .map(function() { return +1 }),

    // reset
    Rx.Observable.fromEvent($('#res'), 'click')
    .map(function() { return -total })
) // merge
.forEach(function(delta) {
    total += delta
    $('#out').text(total)
})

一切都按预期工作.单击+/-递增/递减计数器,然后单击"重置"将其重置为零,但是......我在顶部有变量"total".这是状态,如果我订阅功能反应式编程的值,是邪恶的,不是吗？如果是这样,我该如何解决这个问题？如果我没有重置按钮,我可以使用scan(seed, accumulator),但重置按钮的功能是让我循环,关于如何做"无状态".

在这里工作小提琴.

Answer 1

我可以看到有两种方法可以解决这个问题.

首先,没有任何内容表明您无法在管道中扩充数据:

Rx.Observable.merge(
  // decrement
  Rx.Observable.fromEvent($('#dec'), 'click')
    .map(function() { return {delta : -1}; }),

  // increment
  Rx.Observable.fromEvent($('#inc'), 'click')
    .map(function() { return {delta : +1}; }),

  // reset
  Rx.Observable.fromEvent($('#res'), 'click')
    .map(function() { return {reset : true}; })
) // merge
.scan(0, function(acc, value) {
  return value.reset ? 0 : acc + value.delta;
})
.forEach(function(delta) {
  $('#out').text(delta)
});

上面让你通过在字段中添加来向下游发信号通知已经重置了流(注意:我作弊,为了便于阅读,你可能想要添加reset : false而不是依赖于虚假,但这取决于你).

或者,如果您认为reset实际上正在重置流,那么您可以使用flatMapLatest包装递增和递减:

Rx.Observable.fromEvent($('#res'), 'click')
.startWith(null)
.flatMapLatest(function(e) {
  return Rx.Observable.merge(
    // decrement
    Rx.Observable.fromEvent($('#dec'), 'click')
      .map(function() { return -1 }),

    // increment
    Rx.Observable.fromEvent($('#inc'), 'click')
      .map(function() { return +1 })
  )
  .scan(0, function(acc, delta) { return acc + delta })
  .startWith(0);
})
.subscribe(function(value) {
   $('#out').text(value) 
});

这使得流比包含两个.startWiths来启动相应的序列有点麻烦,但如果你反对扩充并希望由流隐式控制状态,那么这将是一种方法.

Answer 2

在@paulpdaniels回答之后,这是我和Ramda一起使用的:

var hardSet  = Rx.Observable.fromEvent($('#set');
var decRes   = Rx.Observable.fromEvent($('#dec');
var incRes   = Rx.Observable.fromEvent($('#inc');

Rx.Observable.merge(
  incRes.map(function()  { return R.add(1); }),
  decRes.map(function()  { return R.add(-1); }),
  hardSet.map(function() { return R.always(0); })
).scan(function(prev, f) { 
  return f(prev); 
}, 0);