RxJS - make counter with reset stateless?

Question

Assuming I have the following markup:

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

And the following Rx.js script:

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

Everything works as expected. Clicking the +/- increments/decrements the counter, and clicking 'RESET' resets it back to zero, but...I have the variable 'total' at the top. This is state, which if I subscribe to the values of functional reactive programming, is evil, no? If so, how do I remedy this? If I didn't have the reset button, I could just use scan(seed, accumulator), but the functionality of the reset button is throwing me for a loop, as to how to do it 'stateless'.

Working fiddle here.

Answer 1

Here is how it (reset + input for changing step) could be accomplished 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);

markup:

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

Live example on codepen

Answer 2

There are two ways of going about this that I can see.

First, there is nothing that says you can't augment your data in the pipeline:

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

The above lets you signal downstream that the stream has been reset by adding in a field (note: I cheated, for readability you might want to add reset : false rather than rely on falsey-ness, but it's up to you).

Alternatively, if you think of the reset as actually resetting the stream then you could instead use flatMapLatest to wrap the incrementing and decrementing:

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

This makes the stream a little bit messier than it has to be with the inclusion of two .startWiths to kick off the respective sequences, but if you were opposed to augmenting and wanted the state implicitly controlled by the stream then this would be an approach.

Answer 3

Following @paulpdaniels answer, here is what i'm using with 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

@Jrop I like this operator Rx.Observable.when. With this one you can reproduce Bacon.update very easy. This is my code and jsbin example:

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

Also will be better if you look at this Join-calculus, New Release and Joins and this Combining sequences