RxJS - Zähler mit Reset zustandslos machen?

Question

Angenommen, ich habe folgendes Markup:

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

Und die folgende Rx.js Skript:

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

alles wie erwartet funktioniert. Durch Klicken auf das +/- wird der Zähler inkrementiert/dekrementiert, und durch Klicken auf "RESET" wird er auf Null zurückgesetzt, aber ... Ich habe die Variable "total" ganz oben. Das ist ein Zustand, der, wenn ich die Werte der funktionalen reaktiven Programmierung abonniere, böse ist, nein? Wenn ja, wie behebe ich das? Wenn ich die Reset-Taste nicht hätte, könnte ich einfach scan(seed, accumulator) verwenden, aber die Funktionalität der Reset-Taste wirft mich auf eine Schleife, wie man es "zustandslos" macht.

Working fiddle here.

Answer 1

Es gibt zwei Möglichkeiten, dies zu tun, die ich sehen kann.

Erstens gibt es nichts, das sagt, dass Sie nicht Ihre Daten in der Pipeline erweitern können:

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

Die oben können Sie Downstream-Signal, dass der Strom in einem Feld zurückgesetzt wurde (Anmerkung durch Zusatz: I betrogen, für die Lesbarkeit möchten Sie vielleicht hinzufügen reset : false anstatt sich auf Falschheit verlassen, aber es liegt an Ihnen).

Alternativ, wenn Sie die reset denken als tatsächlich den Strom Zurücksetzen dann könnte man stattdessen flatMapLatest verwendet das Inkrementieren und Dekrementieren zu wickeln:

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

Dies macht den Strom ein wenig chaotischer als es sein muss mit dem Einschluss von zwei .startWith s, um die jeweiligen Sequenzen zu starten, aber wenn Sie gegen die Erweiterung waren und den implizit vom Stream kontrollierten Zustand wollten, dann wäre dies ein Ansatz.

Answer 2

@Jrop Ich mag diesen Operator Rx.Observable.when. Mit diesem können Sie Bacon.update sehr einfach reproduzieren. Dies ist mein Code und jsbin Beispiel:

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

wird auch besser sein, wenn Sie diesen Blick auf Join-calculus, New Release and Joins und dieser Combining sequences

Answer 3

Nach @paulpdaniels beantworten, hier ist das, was ich mit Ramda bin mit:

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