throttle debounce with csp channels

debounce-demo.js

/***** demo: debounce *****/

let dIn = chan(buffers.dropping(1)),
  j = 1
let schedulePuts = () => setTimeout(() => putAsync(dIn, j++), j * 10)
let printAndInc = msg => {
  if (msg >= 20) dIn.close()
  console.log(msg)
}
throttle(printAndInc, 500, dIn)
setInterval(schedulePuts, 100)

throttle-demo.js

/***** demo: throttle *****/

let print = msg => console.log(msg)

let tIn = chan(10),
  tOut = chan(1)

for (var i = 1; i <= 10; i++) putAsync(tIn, i)
tIn.close()

setTimeout(() => {
  console.log("-- tOut close --")
  tOut.close()
}, 2500)
throttle(print, 500, tIn, tOut)

throttle.js

let { go, chan, poll, put, putAsync, timeout, NO_VALUE, CLOSED, buffers } = csp

function* readThenCall(cb, inch, delay = 0) {
  let msg = yield inch,
    cancelled
  let cancel = () => {
    cancelled = true
  }
  while (msg !== CLOSED) {
    cb(msg, cancel)
    if (cancelled) break
    if (delay > 0) yield timeout(delay)
    msg = yield inch
  }
}

function* readThenPut(outch, inch, delay = 0) {
  let msg = yield inch,
    cancelled
  while (msg !== CLOSED) {
    cancelled = !(yield put(outch, msg))
    if (cancelled) break
    if (delay > 0) yield timeout(delay)
    msg = yield inch
  }
}

function* readThenPutThrottled(delay, inch, outch) {
  let msg = poll(inch),
    cancelled
  if (msg !== NO_VALUE) cancelled = !(yield put(outch, msg))
  if (!cancelled) go(readThenPut, [outch, inch, delay])
}

function throttle(cb, delay, inch, outch = chan(1)) {
  go(readThenCall, [cb, outch])
  go(readThenPutThrottled, [delay, inch, outch])
  return outch
}