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)
})

一切都按预期进行。单击 +/- 递增/递减计数器，然后单击“RESET”将其重置为零，但是......我在顶部有变量“total”。这就是状态，如果我订阅函数式反应式编程的价值观，那是邪恶的，不是吗？如果是这样，我该如何补救？如果我没有重置按钮，我可以使用scan(seed, accumulator)，但是重置按钮的功能让我陷入了一个循环，至于如何做到“无状态”。

Working fiddle here.

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

@Jrop 我喜欢这个运算符Rx.Observable.when。有了这个，你可以很容易地复制Bacon.update。这是我的代码和jsbin 示例：

const {when, fromEvent} = Rx.Observable;

const decObs = fromEvent(document.getElementById('dec'), 'click');
const incObs = fromEvent(document.getElementById('inc'), 'click');
const resetObs = fromEvent(document.getElementById('res'), 'click');

when(
    decObs.thenDo(_ => prev => prev - 1),
    incObs.thenDo(_ => prev => prev + 1),
    resetObs.thenDo(_ => prev => 0)
).startWith(0).scan((prev, f) => f(prev))
.subscribe(v => document.getElementById('out').innerHTML = v);

如果你看看这个Join-calculus、New Release and Joins和这个Combining sequences也会更好

Answer 3

按照@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);

Answer 4

这是 RxJs v6 的实现方式（重置 + 更改步骤的输入）：

const { fromEvent, merge } = rxjs;
const { map, mapTo, startWith, scan, withLatestFrom } = rxjs.operators;

const resultEl = document.getElementById("js-result"),
  stepInpEl = document.getElementById("js-step-inp"),
  btnDecEl = document.getElementById("js-btn-dec"),
  btnIncEl = document.getElementById("js-btn-inc"),
  btnReset = document.getElementById("js-btn-reset");

// observables (5)
const step$ = fromEvent(stepInpEl, "input").pipe(
  startWith({ target: { value: 5 } }),
  map(e => Number(e.target.value))
);

const inc$ = fromEvent(btnIncEl, "click").pipe(
  withLatestFrom(step$),
  map(([e, step]) => step )
);

const dec$ = fromEvent(btnDecEl, "click").pipe(
  withLatestFrom(step$),
  map(([e, step]) => -step )
);

const reset$ = fromEvent(btnReset, "click").pipe(
  mapTo( 0 )
);

const counter$ = merge(dec$, inc$, reset$).pipe(
  startWith( 0 ),
  scan((acc, value) => value && acc + value)
);

// subscriptions (2)
counter$.subscribe(val => resultEl.innerHTML = val);
step$.subscribe(val => stepInpEl.value = val);

标记：

<h2>RxJS (v6) counter</h2>
<em>5 observables, 2 subscriptions</em>
<h3>Result: <span id="js-result"></span></h3>

<button id="js-btn-reset">Reset</button>

<input type="number" id="js-step-inp" class="stepInp">

<button id="js-btn-dec">Decrement</button>

<button id="js-btn-inc">Increment</button>

codepen 上的实时示例