state-subject-example.js

// source http://codepen.io/yamalight/pen/596767b069e6bc68a95d6d68a2d3979c?editors=001 on RxGitter

const addStream = new Rx.Subject();
const deleteStream = new Rx.Subject();
const modifyStream = new Rx.Subject();

// setup state
const stateStream = Rx.Observable.merge(
        addStream.map(v => state => state.concat(v)),
        deleteStream.map(v => state => state.filter(x => x.id !== v.id)),
        modifyStream.map(v => state => state.map(x => x.id === v.id ? v: x))
    )
    .startWith([])
    .scan((state, f) => {
      console.log('state in scan:', state);
      console.log('f in scan:', f);
      return f(state)
    });


// subscribe to results
const sub = stateStream.subscribe((d) => console.log(d));

// actions
// add
addStream.onNext({id: 1, text: 'test'});
addStream.onNext({id: 2, text: 'test two'});
addStream.onNext({id: 3, text: 'test three'});
addStream.onNext({id: 4, text: 'test four'});
// delete
deleteStream.onNext({id: 2, text: 'test two'});
// modify
modifyStream.onNext({id: 4, text: 'test four changed'});