rx.derivables.js

let R = require("ramda")
let {Observable, Subject} = require("rx")

let scanFn = function (state, updateFn) {
  if (typeof updateFn != "function" || updateFn.length != 1) {
    throw Error("updateFn must be a function with arity 1, got " + updateFn)
  } else {
    return updateFn(state)
  }
}

let store = function (seed, update) {
  return update
    .startWith(seed)
    .scan(scanFn)
    .distinctUntilChanged()
    .shareReplay(1)
}

let derive = function (inputPaths, outputPath, deriveFn) {
  let inputLenses = inputPaths.map((path) => R.lensPath(path.split(".")))
  let outputLens = R.lensPath(outputPath.split("."))
  let inputStreams = inputLenses.map((lens) => {
    return this.map((state) => R.view(lens, state)).distinctUntilChanged()
  })
  let outputStream = Observable.combineLatest(...inputStreams, (...args) => {
    console.log("Calc derived with:", args)
    return deriveFn(...args)
  }).distinctUntilChanged()
  return this.combineLatest(outputStream, (state, stateFragment) => {
    return R.set(outputLens, stateFragment, state)
  })
    .distinctUntilChanged()
    .shareReplay(1)
    .debounce(1) // hide glitches (tradeoff: also hides sync updates)
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

let seeds = {
  irrelevant: "foo",
  input: 1,
  output: null,
}

let updateS = new Subject()
let update = updateS.map(x => x)

let state = derive.call(store(seeds, update), ["input"], "output", (input) => {
  return input * 2
})

// real API is
//   store(seeds, update)
//     ::derive(["sourcePath-A1", "sourcePath-A2"], "targetPath-A", fnA)
//     ::derive(["sourcePath-B1", "sourcePath-B2"], "targetPath-B", fnB)
//     ...

state.subscribe((s) => {
  console.log("Visible state:", s)
})

setTimeout(() => {
  updateS.onNext((s) => R.assoc("input", 2, s)) // expect single "Calc derived", expect `output` to become 4
}, 100)

setTimeout(() => {
  updateS.onNext((s) => R.assoc("irrelevant", "bar", s)) // expect zero "Calc derived", expect `output` to not change
}, 200)

setTimeout(() => {
  updateS.onNext((s) => R.assoc("input", 3, s)) // expect single "Calc derived", expect `output` to become 6
}, 300)